mslot

hashtext() works well

dbt comes to mind

You're responding to said grump ;)

Andres is one of the most prolific PostgreSQL committers and his depth of understanding of systems performance is second to none. I wouldn't have guessed he would one day save the world with it, but there you go.

An obvious answer would be PL/Python, though is not available on most managed services because they prefer to prevent file system access. An exception is https://docs.crunchybridge.com/extensions-and-languages/pyth...

On the other hand, it's quite useful to have a pure SQL cron that cannot do everything, but can run safely in managed services or under a restricted database user. That's probably one of the reasons why practically every managed PostgreSQL service offers pg_cron. Doing more things, like running shell scripts, might limit its usefulness.

Fwiw, pg_cron jobs run as the database user that scheduled the job and the jobs table is not writable, nor can you see jobs created by other users unless you are the admin.

There is not much you can do with it that you could not via external infrastructure that connects to the database, except you don't need to set up external infrastructure that connects to the database. (and simpler systems are generally more secure)

I always wonder why this pattern is not more popular: https://sive.rs/pg

We try to put a clean API on everything, except the one place where it really matters...

One challenge with running PostgreSQL in production on Kubernetes is that it does not behave nicely under high memory pressure. PostgreSQL likes to get NULL back from malloc such that it can gracefully abort transactions (overcommit_memory = 2 in Linux). However, this model of tracking and dealing with memory pressure is still not available at the cgroup level. Instead, processes will be OOM killed. In PostgreSQL, that triggers potentially lengthy crash recovery and therefore downtime. On the flip side, the cost of deploying a hot standby is often lower on Kubernetes, but failing over due to high memory pressure can also lead to pathological behaviour.

A good discussion of this issue by Joe Conway: https://www.crunchydata.com/blog/deep-postgresql-thoughts-th...

Hence, not completely the same as running on a VM.

Note that cstore_fdw ≈ Citus columnar (modernized using table access method) ≈ Azure Cosmos DB for PostgreSQL (managed Citus service) ≈ Hydra (a Citus columnar fork)

Fwiw, the main use case we had in mind when developing the Citus columnar table access method was compression of old data in a time-partitioned table. Citus is commonly used for real-time analytics on time series data. That involves more materialization than typical OLAP reporting use cases. However, you might still want to keep the source data in the database for ad-hoc queries or future materialization and it's preferable to keep the source data in compressed form.

Redshift is specifically architected for ad-hoc OLAP queries. AlloyDB I'm not sure.

> What's advantage over having tenant id as distribution column?

Ease of use is definitely the main one. If you're willing to put in the work required to use tenant ID as a distribution column (add it to tables, primary keys, foreign keys, filters, joins), then it's a more scalable approach.

A challenge with sharding by tenant ID is that many applications use a normalized data model, meaning not all tables obviously have a tenant ID column. When you use a schema per tenant on vanilla PostgreSQL, no additional steps are typically required to enable schema-based sharding.

There are some other benefits of schema-based sharding such as custom table definitions, simpler & more versatile access control, and longer term we expect it will be easier to pin a large tenant to a node using schemas, or distribute the tables of ultra-large tenants (by some other dimension). Of course, row-based sharding has other benefits like parallel cross-tenant queries and global DDL.

> Couldn't the microservice case be handled by having distributed tables with no distribution column?

Absolutely. We first implemented the notion of single shard distributed tables with a NULL distribution column, and then built schema-based sharding on top as a convenience layer.

> Can one have a reference schema which can be efficiently used alongside every other schema? Guess that's public schema with create_reference_table/create_distributed_function

Yes, public schema (or other another non-distributed schema) can have reference tables & distributed tables as usual, and tables in distributed schemas can have foreign keys / local joins with reference tables.

The answer is mostly the same as for PostgreSQL, since Citus stores data in regular PostgreSQL tables.

It's a good idea to set up archival into blob storage / S3 using a tool like WAL-G for disaster recovery purposes, and streaming replication using a tool like Patroni.

(Or use a managed service like Azure Cosmos DB for PostgreSQL)

If a node is down and cannot be quickly recovered then the remaining shards are still available for reads and writes, except for DDLs.

Historically, there has been, but we have found physical replication is ultimately a lot more performant, robust, and tweakable. Citus users & platforms generally use physical replication of each node for high availability.

For instance, the ability to quickly spin up a new replica using a disk snapshot is very useful and only feasible at the server level.

It is still possible to replicate shards (via the citus.shard_replication_factor setting), but it only helps for scaling read throughput, at the cost of lower write throughput.

True, though I'd phrase it more as a middle ground. Which model is best depends on your needs.

A big advantage of schemas over databases is that you can access different schemas over the same connection, which allows much higher density, since connections are a scarce resource in PostgreSQL. You can also have shared tables.

A big advantage over row-based sharding is that you can have variations of the table definitions or a different set of tables altogether in each schema. Plus for the purpose of sharding it's easier to use, since normalized data models complicate row-based sharding.

Nice to see this on HN :)

The high-level is: You enable a setting and every CREATE SCHEMA creates a new shard. All the tables in the schema will be co-located so you can have efficient joins & foreign keys between the tables.

On top of that, you can also have reference tables that are replicated to all nodes, again for fast joins & foreign keys with all schemas.

Everything else is about making every PostgreSQL feature work as seamlessly as if there was no sharding. You can still do things like transactions across schemas, create and use custom types, access controls, work with other extensions, use procedures, etc.

Their lights are nice, but their motion sensor drains the battery in 2 months or so. Eventually I figured out that I could control the IKEA lights via Philips Hue, so I could switch to a Hue sensor which is already lasting a few years.

In general, Philips Hue / Signify is very reliable in my experience, though expensive.

Most $100B+ companies (e.g. Google, Meta, Amazon) were built primarily using relational databases for the first 10-15 years.

Here you go!

https://www.citusdata.com/blog/2022/06/17/citus-11-goes-full...

Not right now. It's definitely something that's on our radar, but it'll take a while to get there.

One thing I forgot to mention in the blog post: Since all nodes in a Citus cluster now look the same, you may be missing your shards (Postgres tables that store a subset of the data in a distributed table).

The shards are still there, but we use some tricks to hide them from PostgreSQL clients since many users get confused by seeing the shards next to their distributed tables. To reveal the shards, you can set citus.override_table_visibility to off.

The release notes have more details: https://www.citusdata.com/updates/v11-0-beta/#shard-visibili...

OLAP stores are relatively fast at answering a single query on a large data set, but basically none of them can handle high throughput with subsecond response times (e.g. when the whole country checks statistics for their own postcode at 4pm).

OLTP stores are relatively bad at aggregating across a lot of data.

Analytics dashboards with many users, a lot of ever-changing data, and many different views exist in a gray area between OLAP and OLTP often referred to as real-time analytics or operational analytics. The queries are usually somewhat lighter / less ad-hoc / more indexed than in OLAP, but there can be hundreds or thousands of them per second with different filters and aggregations.

There are some specialized real-time analytics databases like Druid. Citus (used in the article) allows you to run such workloads at scale on PostgreSQL.

The dashboard has to deal with a complex data integration problem, with different sources with differences in completeness, accuracy, age, and granularity (at many levels), daily corrections in past data, changes in data structure and semantics over time, large data volume, 4pm traffic spikes. Moreover, an API that allows you to select different metrics for different areas. Being able to simply write a SQL query or update and have it be fast regardless of volume is quite a life-saver if you have a tiny (mostly 1 person) team and development speed/adaptability is essential.

Some example queries issued by the dashboard: https://github.com/publichealthengland/coronavirus-dashboard... https://github.com/publichealthengland/coronavirus-dashboard... https://github.com/publichealthengland/coronavirus-dashboard...

Good code explains the problem to the reader and implements the solution as a side effect.

You can often express the problem quite well using a combination of if, for, comments, variable names, function names, list/set and map data structures.

Sometimes (<10%) you encounter a problem that's interesting enough to define a custom type, and sometimes you make tree data structure out of it (<1%). On very rare occasions (<0.1%), you need a more complex data structure.

Code that uses thousands of classes is very hard for a reader to approach, but so is a single file with a never-ending ifs and for loops. Always write code for another person. If nothing else that other person might be you in a few months (or much sooner).

Coding is basically just writing for humans in a language a computer can execute.

True, it is developed by Microsoft and available as a service on Azure. It is also open source, actively maintained and improved, and it's a PG13-compatible Postgres extension that adds both distributed database capabilities and columnar storage. :)

https://github.com/citusdata/citus

(Citus engineer)

It is a custom format that was originally derived from ORC, but is very different at this point. For instance, all the metadata is kept in PostgreSQL catalog tables to make changes transactional.

Correct, though it depends whether you are CPU-bound or I/O-bound. We see the latter a lot more often for large data sets.

Columnar storage for PostgreSQL is especially relevant in cloud environments. Most database servers in the cloud use managed, network-attached disks because of durability, availability, and encryption-at-rest requirements. However, those do come with a performance penalty compared to local SSDs. The VMs also have IOPS and bandwidth limits, partly to manage capacity within the IaaS platform.

If you can reduce the data size by 10x, then you are effectively increasing your disk bandwidth by that much as well. Moreover, you can keep more data in memory, so you will read much less data from disk, plus you'll only read the columns used by the query. Hence, you're likely to see speed ups of more than 10x for some queries, even without column-oriented execution.

It always depends on the data, but we've seen 92.5% and more: https://twitter.com/JeffMealo/status/1368030569557286915

The access method approach followed in Citus is indeed lower level and more generic, which means it can be used on both time series data and other types of data.

For time series data, you can use built-in partitioning in PostgreSQL. It's not as easy to use as TimescaleDB, but pg_partman goes a long way, see: https://docs.citusdata.com/en/latest/use_cases/timeseries.ht...

You can then use the columnar access method to compress old partitions (see the end of the doc), and use distributed tables to shard and parallelize queries and DML.

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)

One caveat to keep in mind when using the binary format is that arrays of custom types are not portable across databases because the serialized array contains the OID of the custom type, which may be different on the other end.

The other thing to keep in mind is that text or CSV can be much more compact for data sets with many small integers or NULLs. On the other hand, the binary format is much more compact for timestamps and floating point numbers. In general, binary format has lower parsing overhead.

The important thing to remember is to use COPY .. FROM STDIN, not insert, for bulk loading into PostgreSQL. Most PostgreSQL drivers support COPY from the client, though it's not always available through generic database APIs.

COPY commands typically write hundreds of thousands of rows per second on a large server/cluster. It's useful to write over multiple connections, but rarely more than ~16.

Author here.

What's kind of unique about the distributed functions feature in Citus that's described in the blog post is that you can scale stored procedures horizontally if procedure calls query co-located shards, but it doesn't restrict you from doing queries and transactions across all shards.

Happy to answer any questions or debate the merits/evils of stored procedures.