Citus + StackGres
Nik: Hello, hello, this is Postgres.FM,
as usual.
I'm Nik, PostgresAI, and as usual
with me Michael, pgMustard.
Hi, Michael.
Michael: Hi, Nik.
Nik: We have a great guest today.
Did we have you already?
Álvaro from Ongres.
Álvaro: Hello.
No, it's my 1st time here.
At least as a speaker, not as a
listener.
Nik: Thank you for coming.
I know you're a very active listener.
We definitely have you on Postgres
TV channel, on YouTube channel,
talking about licenses.
That's why I'm confused.
Do you remember?
Álvaro: Yeah.
Nik: And I know we will touch a little
bit licenses today, just knowing
the topic.
Yeah, thank you for coming.
I'm especially excited that today's
topic will connect so many
dots, right?
Performance, sharding, Kubernetes
operator, what your company
is doing.
It's so open source and true open
source, right?
I'm very excited, looking forward
to dive into this chat because
many topics I'm very interested
in myself and my company also,
they are connected today.
Álvaro: No, thank you very much
for having me.
It's been a lot of episodes listening
with you over the year
and now being present here, it's
very exciting for me.
So thank you.
Nik: And today we will talk about
benchmark you haven't published
yet.
So thank you for coming with results
being still prepared.
That's great because I know we
scheduled this episode before
I think you decided to work on
it.
So this is like...
Álvaro: Right, that's correct.
Nik: Action happening in real time.
That's great.
And yeah, I received your pictures,
3 charts about benchmark
data, benchmark results this morning,
so I'm catching up.
Where should we start actually?
Because this is a great topic,
right?
So how Postgres can scale, what
numbers we can achieve.
Maybe we should start mentioning
that Michael and I had an episode
about our company benchmarks on
1 node, no sharding at all, just
1 node, we achieved 4 million TPS
a couple of years ago.
It was a couple of years ago, right?
With SELECT only primary key
lookup just to see how it's not
super huge compared to some other
database systems, which might
be faster in specific workloads,
but it's great that general
Postgres can achieve that.
And you can give us numbers much
bigger because you involve sharding,
right?
Álvaro: That's correct.
If you want, I can start a bit
with my thesis about sharding
in general for Postgres.
I think it's a field where we're
seeing a lot of innovation happening,
and I love this.
Innovation and competition are
great, are what drives us forward.
Nik: Let me interrupt you with
plugging in episodes with Sugu,
Multigres I think we had 1 or
2, and definitely we had 2 episodes
with Lev PgDog.
So There is an innovation starting
last year, which is great.
And I know you have your own opinion.
Apologies for interrupting.
Let's continue.
Álvaro: Yeah.
Yeah.
I do have my opinion.
Basically databases, they take
a long time to mature.
Databases need to be the places
where you can trust your data,
that it's gonna be there, that
you're durable, and they obviously
obey the rest of the properties
that you expect from a database.
And typically new databases are
risky or tend to lag adoption
because they need to mature.
When you're sharding, my thoughts
keep being the same.
You really need the mature and
trust solution, a boring, I would
say, technology to run your shards
because now you don't have
1 cluster, you have many clusters.
Each shard, each portion of the
database, it's a cluster on its
own.
So the software that manages all
that needs to also be well-established,
let's say.
And again, in competition and innovation
are absolutely great
and very, very welcome.
But I believe that there's a great
technology out there, open
source for sharding Postgres that's
been in the market for, I
guess, like more than 10 years
called Citus.
And Citus is just great.
Might not be perfect, but it's
really good enough or more than
that even.
And a lot of people may have forgotten
about it or put it a little
bit aside.
Myself, the same several years
ago, until I started with thinking,
OK, people need to shard Postgres
and people need to scale Postgres.
And not only because of the performance
directly, because as
you demonstrated, you can run very
high TPS on a single node.
But sometimes it's also about the
blast radius.
Like if you're running a SaaS shop
or an e-commerce and your
database cluster goes down for
whatever reason.
Sometimes it's a configuration
issue with the network, whatever
it is, right?
Your cluster is down, then your
whole thing is down.
If you shard and you split your
database into, let's say 16 logical
shards, right?
Or workers.
And then 1 goes down because of
a configuration error, then it's
only a small percentage of your
users that are effectively down,
which it's not good, but it's definitely
not 100%.
Your PR is completely different
night and day, right?
So a lot of people now need to
shard either because of performance
reasons, because they cannot scale
the writes to a single primary,
and or because of high availability
and blast radius reasons.
And we always had Citus there.
And Citus is fully featured.
It has almost everything that you
need for sharding, for running
complex queries that go down, they're
pushed down to the workers,
and has the tools for rebalancing,
for dealing with hot shards,
for almost all operational activities
that you need to perform
on a sharded database and it just
works.
I believe because their initial
focus was more OLAP style or
mixed OLTP OLAP that people started
considering Citus only for
that and it's definitely good fit
for that but it's also really
good for OLTP.
Nik: 2 points here.
1st of all, you said we always
had.
Let me disagree with you because
when I actually was looking
into Citus, it was several years
ago.
The Key features were not open
source.
For example, online rebalancing.
So we didn't have it in the open
source area.
It was like you need to have a
contract with them to have online
rebalancing, so it quite quickly
becomes out of scope if you
want to rely on open source and
avoid vendor lock-in.
Now it's different.
It has changed a few years ago.
All features after acquisition
by Microsoft, at some point all
features went to open source, which
is great.
But it's not like we always had,
right?
Álvaro: Yeah, you're absolutely
right.
And thanks for that.
I think, if I'm not mistaken, Citus
version 11 was the 1st 1
that included everything as open
source.
And that was, I think, 2022.
So 4 years ago, 4 years and 0.5
ago, that was released as fully
open source and has been since
then.
So yes, you're right.
For this podcast,
Nik: you're right also, because
this podcast exists for years.
Exactly.
So, good.
And the 2nd point is OLAP versus
OLTP.
My concern, like, last 4 years,
I think I didn't dive into Citus.
I remember, diving means like I
run it, inspect myself, the benchmark
and so on.
I did it before we started podcast
actually.
I remember I was reporting some
bugs And since then I didn't
dive into it, but my concern is
we didn't see great examples
of some well-known companies, SaaS
companies, for example, running
it.
Like, you know, if somebody like
GitHub scale would be running
Citus, I wouldn't have any concerns
at all.
What are the great examples?
OLTP style.
That's my concern.
And I think it's still like, I
agree, like what numbers you bring,
and this is interesting.
We dive into them today.
And I was in talks with Citus team
some time ago, maybe a couple
of years ago.
And they presented me some examples
which were not OLTP.
For example, some search for, I
think, lawyers of video, where
I thought like, actually lawyers
can wait 10 seconds to find
problematic materials and so on.
It's not OLTP when we know 200 milliseconds,
our very 1st episode
in this podcast means that humans
already notice delay.
Álvaro: So I think there's 2 angles
to what you're saying.
So 1 is the technical 1, and the
other 1 is the marketing or
trust 1, let's say.
On the trust 1, I would say you're
right, meaning that there's
no such great examples.
And this actually piggybacks on
what I was saying.
It looks like it's a technology
that didn't get the momentum
it could have had, or it could
still has, because of the technological
means that it conveys, right?
I can tell you that we have right
now significant examples of
this kind of usage.
We have now a customer running
Citus on StackGres that is actually
growing database towards 2.2 petabytes
by the end of the year.
And...
Nik: Wait, hold on 1 2nd, 2.2 petabytes?
Yes.
It's just 1 big Citus cluster?
Yes.
Álvaro: Running on StackGres?
Yes.
Nik: Okay.
How many nodes?
Álvaro: 24.
Each node is, I'm not sure if it's
64 or 96 cores, and 0.5 a
terabyte or more of RAM.
Nik: Wow, okay.
Álvaro: Those numbers are actually
quite impressive.
It's an IoT system and they ingest
data massively from millions
and millions of devices that send
heartbeats every 5 minutes.
And they have a 10 day retention
period.
So basically they're deleting 1
10th of the database and ingesting
1 10th of the database every day.
Right now they're in ramping up
period, and as of today, they're
ingesting some 30 plus terabytes
per day, and this is gonna grow
to 200 terabytes per day.
So they're growing by 6X by the
end of the year.
And yeah, this is running on a
single Citus cluster.
We can discuss whether this is
a little bit OLTP or OLAP.
It's a bit of both.
Ingestion is OLTP.
Queries, there are a little bit
of both.
Also OLTP queries when they're
accessing health information about
a given device.
Or more aggregated reports, like
how many devices are down or
having some issues.
So it's a bit of a mixed workload,
but I think it works.
Again, I agree, there are not so
many public great examples about
using Citus.
Citus also introduced some releases
ago, the support for sharding
by tenant, which is actually directly
towards SaaS enterprises,
right?
The SaaS applications where you
don't shard by row, you shard
by tenant, which is a great fit
for those cases too.
Typically, these cases where you
create like a lot of tables
for every tenant.
So it's another way of isolating
those workloads.
But from a technical perspective
and speaking.
Nik: Yeah, I just, I just, I'm,
I'm, I'm smiling at my glasses
because they said, they started
arguing with me saying that there
are public well-known names like
Heap and the Cloudflare named
on, on Citus pages for, for
years.
But, but both of these cases are
analytical cases, 1st of all,
and both these cases went away
from Citus, as we know.
Like, I was at Meetup when Cloudflare
explained how they moved
away from Citus to ClickHouse.
And I know Heap Analytics moved
somewhere else as well.
Heap Analytics was a great case,
but again, analytics.
And Cloudflare, it was analytics,
DNS analytics.
So it's not OLTP.
It can be HTAP.
Yeah, I agree.
And your case is interesting.
It's not what I want to see, like
I said, GitHub or some social
media or something, when there
are many humans working.
Your case is a great case for sharding
Specifically because it
shows exactly where the problem
with the biggest pain of single
cluster is a lot of writes.
This case is highlighting this
particular use case, which is
great.
But I'm still like, my heart is
with social media, SaaS, like
these things, like e-commerce,
online stuff, where a lot of humans
try to deal with some system and
then they want greater responsiveness
of all API calls.
Álvaro: So let's talk about that
because part of my response
also was coming from a technical
angle and from a technical angle
we've been running some benchmarks
and 1 of the questions whether
to say the system is ready for
OLTP or not is the number of concurrent
queries that it can run and the
latency at which those queries
run.
And Citus can run on millions of
transactions per 2nd, even under
1 millisecond p50.
I would call that OLTP.
Nik: I agree here.
If you check very fast SELECTs,
we have benchmarks.
We did it multiple times for actually
all sharding systems, open
source sharding systems we can
reach, including SPQR, Citus,
Multigres actually we also tested,
PgDog, something else.
And Citus, you checked for latency
overhead, off latency, latency
overhead benchmark, simple.
Which like if primary key lookup is
sub millisecond, you don't want
another few milliseconds to be
introduced.
And Citus was actually looking
very well.
It was introducing like 200 microseconds
or so, as I remember.
Álvaro: Yeah, I actually have drilled
down on the CPU consumption
per layer for this kind of queries.
And the Total aggregated CPU consumption
per query, including
the coordinator, including the
worker, including the connection
pooling, excluding the network,
is somewhere in between 300 to
400 microseconds.
Nik: Microseconds, exactly.
Yeah, yeah.
Okay.
Slightly different from what I
remember from our benchmarks but
well the point here is that it
can be very quickly eaten by something
like network complexity right you
won't notice this small overhead
behind something else like
just hops between yeah
Álvaro: the networks that can be
worse than that yeah yeah
Nik: that's I agree yeah
Michael: Álvaro when you query
when you're doing that check are
you going through a coordinator
in Citus?
Because I know now you can query
directly 1 of the workers.
Would that reduce the overhead?
Álvaro: So, yes, the usual way
you wire up a Citus cluster is
you have workers, which are where
you split the data into and
then you have the coordinator,
which acts as a metadata entry
point, but also as a query router.
And so it is the entry point by
default.
It's true that Citus at version
11 introduced a functionality
where you can query also the worker
nodes directly.
There's 2 reasons to do this.
1, if you know from the application
perspective, the sharding
algorithm and you know which shard
to talk directly, which is
probably not a good idea, but technically
sometimes you can do
this.
It is a good idea actually, but
the algorithm is a little bit
internal to Citus, right?
If it would be like clear, exposed
algorithm, then it would be
very good for applications that
want to implement that algorithm,
then talk directly to Shards.
But at least Citus gives you the
opportunity to do that if you
want to do so.
You can target directly the workers.
And if you target directly the
worker, there's no query routing.
It goes directly and resolves the
query locally, if it's local
or fans out to other workers if
required.
But the other reason is to try
to scale the single coordinator.
And I believe this actually also
piggybacks a little bit with
what you were saying, Nikolay, that
Citus maybe was abandoned
by some project before, like Cloudflare.
And Citus can scale workers horizontally
as much as you need.
I was referring to this 24 node
cluster before, and there's been
reports of much larger clusters,
like 60 nodes or more.
And they scale very well.
However, the coordinator remains
only 1.
And if you outrun the coordinator,
then you're hitting a ceiling.
That ceiling is normally quite
high and actually we've measured
that ceiling, but it exists, right?
So that's why Citus introduced
functionality, you can query the
routers directly, and then you
have like essentially multiple
coordinators.
But I don't think that's the way
to go either, because then you're
now contending workloads.
So you have nodes that act both
as query routers, which consumes
CPU, consume resources, notable,
they require connections, and
they spread all the connections
to other nodes, making also connection
math harder.
And they're also serving data and
queries.
So it's a bit of a mixed workload
that I believe is hard to dimension
and do capacity planning around
that and makes everything much
more complicated.
So it's not a perfect solution
either.
Nik: I think we forgot an important
piece.
Why Álvaro at all talking about
Citus, right?
This is a good question.
You're doing something else.
Why are you talking about Citus
at all?
Citus is in the hands of Microsoft.
What happened here?
You developed StackGres, and then
somehow these paths joined,
right?
They met these things and you have
examples.
That's why you mentioned this 2
petabyte example, right?
Can you explain a little bit what's
happening, why you focus
on Citus so much lately?
Álvaro: Well, it's not me, it's
what customers and users request.
I see a growing concern and need
to scale Postgres beyond the
single primary for the reasons
I mentioned before, write limitations
and also blast radius.
And that's why we are also seeing
an explosion of new sharding
technologies for Postgres.
So this is a real need and is driven
by market demand.
We are seeing a lot of people coming
to us asking how we can
shard our database.
And StackGres, it's a platform for
running Postgres, right?
That's it, it's fully featured,
it's really advanced, but it's
a platform for running Postgres.
And so we saw this need and we
saw early that we had to provide
users a way to shard.
And actually we were in a very
privileged position, I would say,
because what we've done in StackGres
with single clusters is
orchestrate and allow you to run
the whole stack of components
that you need to run Postgres,
connection pooling, backups, monitoring,
all that, all automated and all
packaged.
So deploying it is extremely easy
while also following best practices
that we have from our company's
operational experience.
But then we saw that creating a
Citus cluster involved a lot
of manual work because it's no
longer 1 cluster.
Now, if you think about it, and
the model we follow in StackGres
is that every component of the
Citus cluster, the workers, the
coordinator, all of those are a
Postgres cluster in itself, typically
with more than 1 instance also
for high availability purpose,
right?
And they need to be wired up precisely
with the same configuration
parameters across many places or
otherwise it will not behave
as a single cluster.
So it's not only a lot of work
to wire up as let's say a 24,
48 node the Citus cluster but also
to do it right.
And With our automation and orchestration
that we built for a
StackGres, we could do this work.
And now the reality is that if
you use a StackGres and want to
create a sharded cluster with Citus,
by the way, we support all
the other sharding technologies
in StackGres, not only Citus,
even though it's the most used
and the most feature complete,
I would say solution for sharding
that we have.
But with the 13 lines of YAML,
really high level, you can create
a Citus cluster of 60 nodes
in a few minutes.
You can even do from a web console,
Click, click, click, boom,
and you get the sharded cluster.
There's nothing else compared to
this other than the managed
service and Microsoft offers on
Azure.
And that's the gap we wanted to
cover because we saw demand and
it's paying off.
I would say a significant percentage
of our new customers and
users come for sharding.
It is very widespread and it's
a need that few people are focused
on solving, especially using a
technology that is proven and
well maintained as of today like
Citus.
It's a
Michael: good point about Citus
being on Microsoft's platform
and it could because it's AGPL
it means other platforms are
going to really struggle probably
just can't offer it right so
actually it makes sense that the
if the only kind of 10 year
old sharding actually it's maybe
it's not the only it because
it wasn't SPQR quite old as well
as you say they feature complete
sharding solution on Postgres if
that's 10 years old but you
can't get it on any managed services
except for 1, actually,
it makes sense that there's demand
for, okay, how do we self-manage
this?
Nik: We need to revisit the topic
of AGPL misconception.
How to say?
So AGPL allows you to run it on
your managed service, but you
cannot modify it without not publishing
source code.
And right, this is 1st thing.
Let me explain my words and then
Álvaro you can correct me if
I'm wrong.
I'm just curious if I understand
it right.
And then Álvaro is not doing a
managed service at all.
StackGres is just like operator
plus of course there is enterprise
edition with support and so on.
This is like money coming from
support, but it can be installed
anywhere, right?
Like having AGPL components doesn't
hurt at all.
And, And finally, I know StackGres
was on AGPL until lately, right?
Can you elaborate this?
On 1 bit.
Yeah, am I right?
It's not an AGPL anymore?
Álvaro: It's very good insider
information, yes.
So, 1st of all, AGPL.
I'm not a lawyer, okay?
But I love AGPL and GPL or in
general copyleft licenses.
I believe people say they are viral,
that they restrict your
freedom.
I believe it's exactly the opposite.
They provide forward freedom warranties,
meaning that whatever
software is published under a copyleft
license, it is providing
a warranty that will never be closed
source.
It will always remain under such
a license.
So it's a forward warranty for
users of Freedom.
Whereas a license like the Apache
2 or BSD, they can be turned
into proprietary software, and
that software may cease to exist
as an open source software.
That's why I like them.
Now, the reality is that a lot
of people believe that they cannot
be used, for example, as a managed
service, but you put it very
well.
You can perfectly well use an AGPL
software while offering a
managed service.
The only thing that the license
force you to do is that whomever
interacts over the network with
your software need to provide
the same warranties that you got
for that software, which means
publishing the source code.
So if you use Citus, and even with
Citus, it's arguable because
you don't talk directly to Citus.
But even if we do a little more
loose interpretation, or because
I have Citus, then I'm bound by
these terms directly, it would
suffice to publish the source code
of Citus, which is a link
and a disclosure that you use Citus,
right?
Then if you patch Citus and then
you improve Citus, even if you
don't submit upstream, your users
have the right to get those
benefits too, which is why I love
this kind of licenses.
But some lawyers disagree with
me, probably they're right because
they're lawyers, I'm not.
And not many people want to use
AGPL on their managed services.
But in my understanding, limited
understanding, I think that's
not necessarily correct.
Nik: Imagine if Postgres was on
AGPL version 3.
RDS would need to publish all the
changes they made to implement
rds_superuser and so on.
It's simplified as well.
Michael: But this is a really great
point because I think Postgres
has benefited from being on a permissive
license in terms of
how much usage it's got.
And maybe we're seeing the exact
opposite with Citus.
Maybe we're seeing that AGPL is
actually costing in terms
Nik: of usage.
Repelled usage.
Repelled, yeah.
Like Google, for example, have
internal, at least in the past,
they had internal directive that
let's avoid the GPL version
3 software completely.
Álvaro: Yeah, it's a big if.
We don't know the answer.
Most likely, because The market
has this, what I believe it's
a misconception, but this general
wisdom that AGPL or GPL licenses
are this viral and so on, then
yes, it looks like copyleft software
is perceived as unideal and permissive
licenses are winning.
And certainly this is the reason
I don't want to avoid the question,
yes, StackGres, and I can announce
this publicly, it's gonna be
licensed under the Apache 2 license,
instead of AGPL, as much
as I wish to do otherwise, but
the market and the users demand
this, so we're happy to
Nik: do that and we will
Álvaro: do that in the next days
actually.
Nik: While still Citus remains
on AGPL and it's just plugged
in, right?
Álvaro: Yep, we support, in StackGres
we support more than 200
extensions.
They come with a lot of licenses
and it's a user's responsibility
to, to know the license they're,
they're dealing with and their
wording.
We just provide the software and
the platform.
Nik: It's like a big Lego and some
pieces are different colors
and so on, right?
Álvaro: Yeah.
And to complete the picture, I
think I want to go back a little
bit to what you were mentioning,
Michael.
That also relates maybe to 1 of
the reasons, or potentially,
I don't know exactly the reasons,
but why I believe that maybe
some companies abandoned Citus
at some point, which is that Citus,
the workers, where you split the
data, you can, again, you can
add as many as you need, but coordinator
is only 1.
And if you saturate the coordinator,
then you're out of luck,
right?
You can use, as we said, you can
query the nodes directly, the
workers directly, but that it's
complicated to operate and you
still also will get as far as you
can.
This is something we saw some time
ago and decided to do something
about it.
And in the most recent version
of StackGres, StackGres
1.19, we have introduced something
called query routers.
Query routers does nothing that
we have actually patched on Citus
or anything like that.
It's unmodified Citus.
It's just an architectural pattern
that we identified that we
could apply to Citus to allow to
have essentially multiple coordinators,
put that simple.
So a query router is like a coordinator,
so it can route queries.
It doesn't have data, it just route
queries.
The only thing it cannot do is
the DDL operations.
The DDL operations go through the
coordinator.
But now you can add as many query
routers as you want.
And so the limitation that Citus
may have had until now with
a single coordinator, it's gone.
And now you can scale by adding
as many query routers as you
want and scale, not only now the
worker side, the layer where
it keeps the data, but also the
router layer to almost any size.
I'm publishing some benchmarks
that we have been running in the
last days about this.
And ballpark numbers, 64 core workers
running simple queries
indexed by primary key, but not
fully in memory, but with a high
hit ratio.
A single coordinator with 3 workers,
the coordinator being also
64 cores, was maxing out at around
550, 000 transactions per
2nd with a pretty good latency,
like under 1 millisecond.
But that's it.
More than 500000, a little bit
over 500000, and the coordinator
is saturated.
The reason why this number is because
these queries are quite
fast to process, and the proportion
of overhead per query processing
is maxed out, right?
It's a lot of overhead per very
small query, per tiny query.
Of course, if you hit all the queries
and they involve I/O.
And reading work, you can scale
further before hitting the limit
of the coordinator.
So we designed this benchmark to
actually find the limit as soon
as possible of the coordinator
at still very high rates, right?
More than 500000 transactions per
2nd.
So coordinator is maxed out.
Then we add 2 coordinators with
the same 3 shards and then 3
coordinators with the same 3 workers,
sorry, the data nodes.
And the performance basically scaled
almost linear.
So from 550, it went to 1, 050,
000 and then to 1, 500, 000.
So pretty much linear growth.
Michael: When you say 3 coordinators,
Do you mean 3 query routers
in the new?
Yeah.
OK.
OK.
Álvaro: Yeah, sorry.
I use the terms a little bit interchangeably,
because for me,
they're conceptually the same,
even though they're not.
You're right.
It's query routers, yes.
So 3 query routers.
The coordinator remains there.
And actually, I am now recommending
that if you use this functionality
with the StackGres, with the query
routers, that you keep the
coordinator small configuration,
small number of cores, just
to run the DDL through there, and
do not use a coordinator for
query routing.
And then use a layer of query routers.
You can start with 1 if you want
and then scale them up as your
needs grow, but they're already
prepared.
You already, they use a different
service in Kubernetes, so it's
just a way to route the queries
and entry point.
And then you already have that
established.
Your application points to the
service that will balance across
all the query routers you may have.
And then it's just a matter of
changing 1 character in your YAML
file, say, query routers 1, query
routers 2, and you would scale
automatically.
Michael: And You mentioned a while
back that StackGres supports
multiple ways of sharding.
Could I use query routers with
any of them or is this a Citus
specific thing?
Álvaro: It's a very good question.
This is Citus specific.
So if we look at the internals,
Again, this is not, we didn't
touch Citus source code.
We could have if we needed, but
it was not needed.
The building blocks were surprisingly
there.
Actually, what I'm surprised that
nobody saw this before.
For us, it was crystal clear because
it all started with the
concept that every worker can function
as a query router, as
a coordinator, right?
We just call them query routers.
So that was already working there.
The way this works is because they
implement the metadata that
Citus replicates across all nodes.
So you know whom to talk to, essentially.
And there's another property, which
is what we combined into
this pattern that we call the query
router.
There is a property that is called
shouldhaveshards, and you
can set it to false.
Shards are the way that Citus internally
distributes the data,
right?
So also to clear, to make the context
clear for everybody.
On Citus, there is workers, which
let's say they are the physical
nodes that contain the data, physical,
virtual, you name it,
the nodes that contain the data.
But the data in reality is split
into shards.
And then you map shards to workers,
at least 1, but could be
more than 1.
Sometimes some other people call
this virtual shards versus real
shards, right?
Citus terminology is workers,
like the physical thing, and shards
like the virtual thing.
And typically we recommend a higher
number of shards to worker
because it helps you also scale
when you add more workers, you
just move some of the shards to
other workers.
You don't need to re-shard the
whole thing.
Right.
It's much, much better pattern.
Actually, by the way, I'm diving
maybe a little bit too deep
into that topic, but I use the
magic number 120.
Most people use powers of 2.
I think that's a mistake, because
then you can only grow by powers
of 2, which by the growth of powers
of 2, can mean to grow a
lot when maybe you don't need to
grow that much.
Versus having 120 is a number that
has a lot of divisors, and
so you can grow from 30 to 40,
for example.
You don't need to go from 30 to
60, you need to double because
they're still evenly divided across
all your workers.
If you take a worker that can act
as a coordinator or a router,
and you set this property
shouldhaveshards to false, then
Citus will not place any shards
on it, it will be data-less.
So essentially a data-less worker,
and that's a good router.
So the pattern was just there.
We just enabled it through architecture
and automation in StackGres.
Michael: Simple, I like it.
Álvaro: So of course, there's more
meat to it when you look into
the internal details, but conceptually
this is it.
Just
Nik: don't write proxy, just use
Postgres.
In yet another area.
Yeah, it's interesting.
Speaking of benchmarks, these numbers,
what concerns me that
we talk about only read-only queries,
I would like to see the
series 2 of benchmarks talking
about writes, because as we already
discussed, the case you have in
hand is 2.2 petabytes.
It's about writes, not reads, because
scaling reads is less interesting,
less challenging, right?
Álvaro: It is actually challenging
And again, I will not claim
that Citus is perfect.
And it is really good, but it's
not perfect.
I guess nothing is perfect.
And it requires tuning.
I'm not gonna say this worked out
of the box.
No, it requires tuning.
And 1 of The hardest part of Citus
actually is connection tuning,
because when you look at Citus,
at any distributed database for
that matter, any sharded database,
it's no longer a single thing,
right?
Now you're dealing with connections
from the client to the coordinator
or the query routers, and from
them to the workers.
And sometimes these queries actually
have a big fan-out.
For example, on this particular
case of the 2.2 petabyte database,
I just cannot mention the customer
name, but it's a big telco.
They are using COPY to ingest data.
And that's amazing because it's
the most efficient way to ingest
data into Postgres.
And they're using 120 shards on
24 nodes, which divides evenly.
Now, if you do a COPY of 50, 000
records, a single connection,
so a single connection tries to
ingest 50, 000 records via a
COPY, that COPY is going to create
24 sub copies.
Because by law of probability,
all 24 shards are going to receive
data.
1/24th of the 50, 000 records.
Because the way that data is sharded
is by taking a hash of the
primary key, and that hash is a
random, let's say, it's randomly
distributed, so it's going to touch
all the shards.
So 1 connection from 1 client now
will create 24 connections.
Citus implements this very well,
just do 24 sub copies.
But you can imagine the potential
multiplication of connections
that may happen here under high
concurrency.
But by applying proper tuning and
doing proper connection management,
this works very well.
Nik: What is proper here?
Álvaro: Oh, well, it's not an easy
exercise.
The connections from the coordinator
or query routers to the
worker nodes, they're bound by
several parameters.
Essentially, Citus has some kind
of, let's say, connection pooling
system to communicate between the
coordinator or coordinator
to the workers.
But it's not like PgBouncer, because
they open backends that
are Citus is Postgres, for the
good and the bad.
There's many good things and a
few bad things.
And so if there's a client talking
to the coordinator, that creates
a backend on the coordinator, right?
And that backend needs to be open.
Now, this backend may now own multiple
sockets to multiple connections
to the workers.
And there's several parameters
to tune whether you want a single
connection per worker, whether
you want a number of connections
per worker.
You can cache connections already
open per worker, but then there's
also like a global pool so that
you don't want to do more than
X number of connections for a given
coordinator across all the
workers.
And so you need to really understand
that we all know that Postgres
doesn't work well with a huge number
of connections because it
creates a lot of backend.
And so you need to understand the
fan-in rules, the fan-out rules,
whether to put connection pool,
for example, in front of the
coordinator or not, and so on.
So the math ends up being non-trivial
sometimes, especially if
you're talking about huge workloads.
Like for small workloads, even
if you're starting with again,
maybe hundreds of thousand transactions
per 2nd, maybe more or
less trivial, like any server,
modern server will deal comfortably
with that.
But when you grow to these sizes,
you really need to size everything
around this.
But what's actually very fun of
this particular case is that
as of today, on this particular
case, we'll need to blog about
this.
Right now, the database is perceived
as slow, but they're running
some software that I think you're
familiar with called pg_ash,
and it's reporting all the time,
waits are mostly ClientRead.
They don't have an ingestion pipeline
fast enough to saturate
the database.
The database is faster than their
ingestion pipeline.
Because again, COPY is really optimized
in Postgres and can
ingest huge amounts of data.
And we have here 24 shards, really
beefy servers that can ingest
data at really, really high speeds.
We've seen, don't remember the
numbers, but multiple gigabytes
per 2nd being ingested.
And the database is just fine.
But the ingestion pipeline is not.
Nik: That's interesting.
So I see, I hear 3 different directions
here.
1st of all, this benchmark in your
blog post, which is coming,
I guess you will publish all the
details as usual, right?
Álvaro: Yes, of course.
Nik: This is our company's common
value, we share it a lot.
Álvaro: Fully reproducible, StackGres
being open source, we'll
open source all the code.
Nik: It's Postgres's open source
and benchmarks should be open
so anyone could repeat them.
Yes.
Yeah, so that's great.
SELECT only because this is to
compare with different benchmark
which is not so open, right, and
not about open source.
That's great.
And there, I guess, this is interesting
to continue exploring
the limits.
And I think maybe it will also...
I expect more benchmarks to be
done because we have multiple
systems competing in purely open
source areas.
Some of them are AGPL, some of
them are Apache licenses, we mentioned
already.
It's interesting to start comparing
them.
It's great that it's happening
and this is maybe the beginning
of some comparison.
It's time to start comparing.
Right now your bet is fully on
Citus, like almost fully.
I hear you that StackGres supports
other sharding as well.
And I guess at some point you can
start supporting those new
alternatives as well.
Why not?
It's just a Lego style thing, right?
This is 1 direction.
Read-only benchmarks, which I guess
it's not so exciting.
2nd would be probably writing benchmarks
like COPY.
When you say COPY, you mention
specific production case, but
it would be great also to have
some benchmark showing, to start
playing with and showing capabilities.
Do you plan to have this benchmark?
Álvaro: We could do that, but I
believe we will start by publishing
some, the data that we can publish
about 1 particular case that
we have already, like this 1, because
I think that already shows
a lot.
And so to be honest, we have
focused quite a bit on benchmarks.
So let me be clear, we've done
2 different benchmarks.
Both are going to be published
and will be fully reproducible
back by code, right?
And so full transparency on everything.
The 1st 1 is a benchmark to demonstrate
the ceiling of the single
coordinator and to see how we can
scale with query routers, right?
Like I just mentioned the numbers,
it can go from 0.5, essentially
500000 to 1.5 million, so almost
linear scalability by going
from 1 coordinator to 3 query routers,
right?
But then we saw actually that for
this particular workload, like
very, very small primary key queries,
mostly in memory with a
non-uniform distribution, that
you end up needing a similar number
of cores on the routers layer than
on the shards or the, sorry,
the workers layer, right?
And so we did some tuning and around
this number that the ideal
is to have 6 routers of 32 cores
each and 3 workers of 64 cores
each.
So that's the same number of cores
on both layers, but 6 routers
and 3 workers.
And we call this a unit of a scaling.
That package, let's say it's well
tuned so that both layers are
close to saturation but before
saturation.
So they operate at the operational
regime and all math about
connection, tuning connection,
pooling, and all that is well
tuned.
And that's basically delivering
1.5 million transactions per
2nd, that unit of a scale.
And then we decided to multiply
this unit of the scale by 2,
by 4, and by 8.
And this is a 2nd benchmark that
we'll be publishing that also
compares to some data points of
another sharding solution, which
has achieved those numbers and
more actually.
And it also shows really good behavior,
like basically to share
the numbers by multiplying this
by 8x, it passes over 10 million
transactions per 2nd.
With 1.something millisecond p50
latency and 6 milliseconds p99,
at 10 million transactions per
2nd.
To put this into perspective, who
needed to do 10 million transactions
per 2nd?
Probably just a few sites on earth
do more than that.
Nik: So it's...
Maybe more.
We have gigantic workloads these
days.
I think here I disagree.
A lot of more.
We see crazy startups which skyrocket
these days.
And their users are agents.
And they, like, just...
Álvaro: Yeah, but most of the time
what I see is that agents
are small databases.
There are many of them, but a small...
We're talking about single cluster,
right?
But anyway, my point is that anyway,
reaching 10 million transactions
per 2nd with very decent latency
numbers, right?
It's an excellent result.
It's home to most of the clusters
we've ever seen.
And so it's a valid solution for
almost anyone.
And That's actually my takeaway,
right?
It can scale almost as much or
as much as you want with really
decent numbers on this pure OLTP.
Nik: So I have 4 directions already.
So you will publish.
I know you will publish this.
You will probably will iterate
this.
I guess this benchmark also not
cheap to run, right?
You need to provision those resources.
Álvaro: Yeah, exactly.
But this is a good point.
We're still a bootstrapped startup,
and we strive to be, we have
to be, very cost efficient.
So running this benchmark, I don't
have the final data, but it's
very approximate.
We've spent around $5, 000 to $6,
000 on running this benchmark,
scaling to 10 million transactions
per 2nd.
Nik: It's AWS or where?
Álvaro: Yep.
Nik: And you just use Spot because
it's not reliable in this
case.
Álvaro: Yeah, it's hard because,
and we had some problems when
you deal with such a large number
of instances.
So on the 10 million transactions
per 2nd, this involved 72 instances
plus the drivers for the workload,
so like around 90 instances.
Just 1 instance behaving wrongly?
Nik: The probability of failure
is too high.
Álvaro: Yeah.
So, yeah, Spot would have most
likely had mean that the test
needs to be repeated over and over
again.
Nik: I see.
That's great.
And that's great cost-wise as well
for 10 million.
So once again, 1 direction is SELECT
only primary key lookup,
trivial workload, I would say.
And I think this is very adjacent
to our benchmarks we did multiple
times for latency overhead is very
in this area, like very fast
queries.
2nd, I still think at some point
there should be some benchmarks
for INSERTs or write workloads,
right?
3rd thing is that you mentioned
many connections is a trouble.
I think, I know Andres Freund did
a lot of great work in Postgres
13 and especially I think 14 to
improve the situation when you
have thousands of connections.
Because before that, thousands
of idle connections affected performance
so badly.
Now it's not so bad.
What does it mean still?
We don't have threads, right?
What's happening if you have 10,
000 connections on single Postgres?
I think this research is very interesting.
I'm talking for myself because
we have a lot of interesting directions.
We have benchmarks.
Our focus mostly on single node
clusters.
Our company is dealing with some
self-implementation, like application
side sharding.
But it's so interesting, I see
some interesting directions here
to benchmark further.
And finally, I'm thinking if you
stop at 10 million, I'm going
to explore the possibility to have
10 million TPS primary key
lookups for a single node.
This probably will encourage you
to go to 50 million or something,
spend a few thousand dollars.
Because if, yeah, I will put it
to midterm to-do list.
If we achieve that, you will be
forced
Álvaro: to cover.
But you know, if you want to do
that, you know, you only have
2 weeks, right?
Nik: No, no, no, no, no.
Go ahead, publish your results.
This is an interesting, friendly
competition and I think some
systems will need it to join.
What I like about this is that
data is shared.
It's about open source components.
Everything is open source.
Data is open source.
Benchmarks are open source.
They can be revisited.
This is great.
This is very good style of competition.
Everyone benefits from, right?
So I'm not going to chase you right
now.
I'm just saying maybe later.
Álvaro: No, I'm very open if anyone
wants to sponsor workloads
to be tested on Citus and see how
Citus can scale to from 10
to 50 million transactions per
2nd or to read-write workloads We
would be more than happy to help
in that regards.
Michael: So yeah.
I think it's, I think what, from
1 of the charts that you sent
over, it's quite an interesting
chart of how the throughput changes
as you scale up.
So I think if you, if you, you
mentioned these units of work,
if, if you add a 2nd unit of work
and the TPS exactly doubles
if it's linear like as you add
more units it's very easy to extrapolate
that of course it can get to 50
I just need 5 times more but
if it isn't if it isn't quite efficient
if you lose 10 each time
if it if it feels like at doubling
the instance size only 1.5x
is the TPS, then it's less believable
that it can scale infinitely.
What kind of numbers were you seeing
there?
Álvaro: So that's a really good
question.
Scaling is slightly sublinear,
with most systems are.
And on our scaling from 1 to 8,
it was 1.94 every time we doubled.
Nik: Okay.
Yeah.
There is a big challenge of scaling
client in this case.
Yes.
That's it.
I mean, I
Álvaro: can go very deep into this
topic.
I guess it's not the moment.
But it's a really, really interesting
capacity planning we had
to do to achieve these numbers.
Again, it sounds like, oh, just
put more nodes and yeah, you
scale 10 million, you just need
to hit enter, right?
No, it wasn't that case.
We had to do a lot of tuning.
The number of connections is actually
mind-blowing because even
though these queries are really
fast and take hundreds of microseconds
to run in reality across all the
fleet, across the CPU aggregated
all the fleet, right?
The number of connections that
you need, the sessions that you
need to drive this is very, very
large.
It's many, many thousands of sessions.
And then they multiply because
if each query router is, I think
where our optimal regime was around
400 connections per query
router on 32 cores, which by the
way, if you look at the numbers,
it's always like the usual rule
for tuning Postgres, right?
The maximum number of connections
is a multiple of the number
of cores.
In this case, the multiple is around
7 because they're very,
very CPU intensive queries because
they almost not touch disk,
right?
So the multiple is a bit high.
Anyway, do you think of 400 connections
coming to each query
router of which there are 48 on
the 10 million transactions per
2nd run, right?
And now each 1 is gonna have outgoing
the same number of connections
in the same number of sessions,
because they're all, but they
all go to all shards.
So the incoming connections at
the shards are the whole fleet.
So you're seeing the 400 multiplied
by 48.
Those are incoming connections
to every worker.
So to actually deal with this,
we had to introduce connection
pooling before the workers.
And it's also been an exercise
because StackGres by default uses
PgBouncer, and PgBouncer is single-threaded.
So were we saturating 1 thread?
Definitely.
So actually this benchmark required,
if I'm not mistaken, 15
PgBouncers per worker, running
with a SO_REUSEPORT functionality
and all handled by StackGres as
a custom container.
So it worked beautifully well,
but you need to do all the dimensioning
correctly, otherwise it will not
achieve these numbers either.
Michael: Did you say 15 PgBouncers
per Worker?
Álvaro: Yep.
Michael: Oh my goodness.
Okay.
Okay.
Álvaro: Good.
Because we're dealing with, I think,
19, 000 incoming connections.
And they're all sending queries
at a really high pace.
Nik: What would it take to reach
100 million TPS?
Álvaro: Well, there's 2 things.
1 is just to keep multiplying.
I think this unit of scale is very
well tuned as of now, and
it just keep multiplying.
As of now, it should be an easy
exercise, other than money-wise.
But I believe the scalability is
gonna start going, is gonna
start going less and less over
time.
And this is, again, because of
how Citus works.
As I said, it's not perfect.
You may need to question whether
you need more than 10 million
transactions per 2nd.
But it's, so Citus is gonna create
a backend per session, per
router.
And this, again, we were just discussing
this, is gonna create,
that multiplied by the number of
routers will become the number
of incoming connections to each
worker node.
So instead of having 19, 200 connections,
if we do this 10 times,
that will require almost 200, 000
connections incoming per node.
Not all of them are active.
Are active actually are that number
divided by the number of
workers, but still a very high
number, a very, very, very fun
difference between the fan-in,
the number of connections incoming
to the workers through the poolers
versus the number of outgoing
connections, the pool connections
to Postgres, which needs to
remain constant at around 600 in
the case of the workers or 800,
something like this, because they
have double the CPUs, right?
And this difference between this
fan-in and fan-out creates inefficiency
in the workers themselves.
So it will start saturating earlier.
They also create inefficiencies
in the routers because they need
to talk to more nodes.
And so, will it reach there?
I don't know.
If someone wants to pour the money
into it, we can test it out.
I believe it may reach to a very
high number, maybe with a slightly
less efficiency, but it will reach
maybe 50 million, maybe more
with more effort, but it will do
so.
It's still compared to other sharding
solutions that can reach
those numbers.
I don't think all can, but out
of those that we have tested,
it doing so, this 10 million transactions
per 2nd, is using 0.5
of the silicon compared to other
sharding solutions.
So even though the scaling may
be slightly less linear because
of this Citus connection model,
which is Postgres connection
model, it needed 0.5 of the cores
to achieve those results.
So it's still very impressive,
right?
Now, Do I see a future different
than this?
I do.
I believe there may be an option
where the transport that Citus
uses could benefit from a multiplexing
transport, something like
gRPC, which other of the modern
solutions are adopting.
And that significantly reduces
the fan-in and fan-out connection
problem from the routers to the
workers.
So if I were to complete my initial
thesis that Citus is a great
solution as of today that is very
mature, well proven, featureful,
and so on, and that new solutions
may take time to mature.
Maybe, maybe, and it's a big if,
by just improving a bit Citus
architecture, it could take sharding
even to the next level if
needed, given that already it reaches
levels that are more than
enough for mostly everybody, but
even for those who may need
an even bigger scale, then maybe
It's just a matter of improving
Citus.
Nik: Yeah, I have only 1 question
to ask in the end.
I know you're comparing to other
benchmarks, and I'm just curious
in terms of, I saw some pictures
you shared with us before this
recording that was involving money,
like some dozens of thousands
of dollars to have this cluster.
I'm very curious, do you compare
with humans involved with support
or just pure open source and plus
cloud costs and that's it?
Álvaro: Yeah, so I did 2 comparisons
and they will be clearly
identified in the blog post that
will announce these benchmarks.
And so there's the pure infrastructure
cost, which is silicon,
like number of CPUs, where essentially
it's like around 55%.
So almost 0.5 the cost in terms
of CPU.
But then we can compute the total
cost to be very clear.
And there's nothing here to hide.
I'm comparing right now with a
benchmark that was run and published
by Neki.
And Neki is a managed service,
and they publish their list price.
And their list price is not the
instance cost.
It is the instance cost plus their
managing fee or their margin.
And when comparing with that, the
cost is not 0.5.
It's depending on what you consider
is between 0.333 or 0.2.
Actually, a well-dimensioned system
should be just 5 times cheaper.
Basically, 0.333 is if you compare
on-demand cost, which nobody
should run like a production cluster
sharded on on-demand cost.
And if you use a basic saving plans,
then you'll be 5 times cheaper.
Yeah.
And this is obviously a managed
service to a software comparison.
You may discuss that there may
be humans involved in the total
TCO calculation.
But my take is that a managed service
and software that does
a managed service for you is not
so much different.
Actually, I call StackGres a managed
software, a managed database
software, because when you say
a managed service, there's no
human typing, right?
There may be, if there's a support
ticket or something happens,
but the operation is just software.
So it's essentially the same for
me.
Nik: I know some humans can send
kill -9 in very big companies offering
managed services sometimes.
That kind of support is not welcome.
Not welcome, right?
kill -9 from support.
Michael: I also think these machines
are so big, like we're talking
about sharding and we're talking
about multiple huge machines
really, like they're expensive.
Often you don't pay this.
If you're actually a company,
Nik: saving
Michael: those machines, you would,
you would have like savings
plans either through PlanetScale
and or bring your own cloud
type thing.
Nik: So you're comparing not apples
to apples, but the numbers
you provided on your side, it could
be less with savings plan
if you pay in advance some number
of years to AWS.
On the other hand, you need probably
to roughly multiply it by
2, thinking about humans and the
support needed to surround this
solution, right?
Very roughly.
Álvaro: Yeah, so that's what I
said.
And so 1st of all, that's why I
want to give both numbers.
So you can 1st of all, compare
from the pure Silicon perspective
to pure architecture perspective,
and then from the sale price.
But as I said, For me, a managed
software, it's just software.
A managed service is just software,
right?
And if the software automates and
runs everything for you, which
is the case in both cases, right?
Because all managed services, there's
no humans doing operations,
It's only software.
So humans only intervene in the
case of issues.
And in those cases, yes.
And you can get like a, we sell
enterprise, the StackGres Enterprise,
which is exactly that.
Support for those cases, right?
And that's something optional,
but that's something you would
add on top of that.
And then the calculation will not
be 5 times cheaper, it will
be
Nik: maybe, I
Álvaro: don't have the numbers
exactly, 3 times cheaper, 4 times
cheaper, but still obviously much
cheaper.
Nik: Yeah, thank you for this clarification,
that's interesting.
Álvaro: And sorry, also to add,
and that's because talking about
cloud, but if you talk about on-prem,
which is definitely in
the space where we also, the StackGres
works anywhere on-prem,
then the math is even much more
into your favor.
Nik: Right, if you own hardware.
If you own
Álvaro: hardware and data centers,
Nik: then it's open source, you
can troubleshoot with your engineering
team if needed.
That's
Álvaro: great.
Thank
Nik: you so much.
Very interesting.
And also I see several paths for
future work.
Maybe we should have another episode
at some point in the future
to revisit the topics in some directions.
Álvaro: Any time.
I would be more than happy to come
back.
Nik: Thank you so much.
It was interesting.
Very interesting to me.
Álvaro: Thank you for inviting
me.