I'm Mattias, I work at PingCap.
We are doing a distributed SQL database called the TIDB.
It's MySQL compatible, so for the clients it just looks the same, and I do have a short
talk about online skimages at scale in TIDB.
Similar to MySQL, a distributed database is slightly different.
MySQL does a metadata lock, so it basically needs to stop the world, no transaction can
go through the metadata lock just to change the metadata.
That means that it's a short lock when you do any kind of DDL, but it also means that
when you're doing replication, this metadata lock actually stops replication a bit, so
if it's not an instant DDL, you would start getting replication delay when the DDL goes
through.
So a distributed database is of course different.
From the client perspective, you should just see a normal database.
You should just expect it to be transactional, it should be acid compliant query with your
normal SQL queries.
For the user, you shouldn't see any changes, but of course underneath it's distributed
on multiple nodes, et cetera.
So if you take ad index as an example of a DDL, we can't do the synchronous stop the
world scenario with the MDL example in MySQL, so that's something that we need to solve.
During that, we do need to copy and create all the index entries for creating a new index
while normal traffic comes in.
So in the beginning, MySQL did more or less stop the full table and copied everything over
and then it released it again.
Nowadays, they are much better on the online and only keeping the metadata lock.
But that's something we need to do better in a distributed database.
So the proposed solution is to version the schema, so every change you're doing to a
schema or a table, you do that as a specific version.
You need to allow sessions to use either the most up-to-date version to the current version
of the schema or a previous version.
So then you can do transitions in between these states or versions.
And we need to guarantee that the states between the previous version and the current version
are compatible.
So that basically means we need to create some kind of states from the before or the
start states to the public state where it's usable.
And I think it's easiest to go backwards to actually see what kind of states are needed.
So here the VN, it's the current version.
And we start by the public, it's the end state, everyone sees the index.
So selects goes there, insert updates, everything goes there.
The previous version, we can actually remove the selects, but it still needs to do all
the updates, insert some deletes there.
And as you see, regardless if you're, so the time here can be a bit confusing.
So the transactions are of course using the real time, current time, but it might see
a different version of the schema because we can't require all transactions to constantly
check for the new schema and stop the world for that.
Let's then move and say we are in this write only state.
Before state for that, then of course we cannot do selects.
Do we need to do inserts before to make them compatible?
Well, we don't actually serve the reads.
So we do not need to do the inserts in this state.
Backfill will help with it.
And then of course comes the question, how would backfill handle it?
So for backfill to handle this correctly, we actually need to have another state between
public and the write only state.
And as you see, statewide for transactions, it doesn't really change anything, but it
gives time for doing this backfilling because when we enter the write reorganization state,
then we know that the state before, it's the write only.
So all changes will be double write.
That means that updates can be a bit tricky because we say we're not doing insert, but
how do we handle updates?
So we need to go a bit deeper to see how that's handled.
In the add index example, we do have the table data that's public.
So everyone should be able to read directly from the table without the index.
So let's say we have at time zero, a session that sees this new write only state.
And it does an insert.
It inserts into the table, and it updates the index.
So you can find the row through the index.
Then later on, another session comes in, but that session has not yet transitioned to this
write only state.
So it's in the state before, and it wants to update this.
So it goes to the table and updates the row.
That's public, so that's what it needs to do.
But then how about the index?
We don't actually need to insert into the index here because that will be handled by
the backfill in the write organization state.
But the trick here is that we actually need to remove the old entry as a part of the update.
So update actually means that we need to propagate the deletes into this new index object, but
we do not need to do the inserts.
So we need to propagate updates as delete only.
And that also makes it easy to handle the delete, so we do need to handle deletes in
the new index.
That also gives a name for the state, so delete only state.
When you're reading this, it's inspired by a paper from Google about online asynchronous
schema changes in F1, so on top of Spanner.
Then it takes some time before you understand exactly why you do need a delete state.
But this is the reason, so we'll be able to move through the different stages.
I'm not inserting the new row in the index or the new entry in the index.
Does that not mean that nothing else in the system can use it because you have to wait
for the backfill to complete?
Yeah, so you don't read from the index until it goes public.
It should complete, okay, so you have to wait for it and it doesn't mess around the way.
It just delays that.
You could have done it at the same time while you're all deleting.
So since if you would insert it, then it would more or less be overwritten by the reorg
phase anyway because the reorg needs to read from a snapshot and take all that data.
So a snapshot taken somewhere when everything were on the right only.
So it would just be overhead of doing the insert.
It wouldn't actually mess up anything.
It would still be correct, but it would just be unnecessary.
And then if we move on from the delete only state, the previous version can actually be
the start state because as long as deletes are done, the previous version does not need
to do anything that really states.
So there we have the different states that it needs to transition through for keeping
transactions running without being blocked.
So here we do have the full part of the asynchronous DDL in online, that's done online in a distributed
database.
Do you support distributed transactions and if you do, what transactions in XA prepare
state?
So we do not support XA transactions right now, but of course if you're connected to
different SQL nodes, it looks just like it is a master or a primary wherever.
So full read and write in however you connect.
So transaction is a bit slightly different.
You cannot have transactions spanning more than two versions.
So you need to either wait or you need to block, stop and fail transactions that are
too long-running.
Okay.
And these versions, you have like several versions associated with a single or nice
game of change.
Yes.
So a single DDL goes through multiple stages.
And currently I'm actually working with partitioning and for alter table reorganize partition where
you take one set of partitions into a new set, then there's another thing.
So during the reorganized phase when you're copying data, you do select from the old one
then you go to public, so you select from this one, which means that if someone is actually
on the right reorganization state, then they will select from that that's not updated in
this one.
So you need to add an additional state between the right reorganization and the public state
just for moving the select.
So it's a double right while moving the reads.
And all this is done in tidy B and I'm not sure how many is familiar with tidy B.
Okay.
Good.
Then let's do a quick introduction to this tidy B is mainly architecture around three
different components.
You have PD which stands for placement driver.
It creates the timestamps for transaction handling and it knows about the data locations.
So it knows where the date on which node the data are.
Then we have an SQL layer that is stateless.
So it's very easy to spin up or scale in the different number of nodes.
Here we have re-implemented the MySQL protocol.
So this is actually written in Go.
And all of it is in Apache 2 license.
So we do not share any code from MySQL or Maria.
It's completely new since 2015 when the project started.
And then we have a storage layer.
The base storage layer is a Thai KV, so it's a distributed key value store.
We even have people that run stats as a distributed key value store and don't bother about the
SQL part.
So that's what you can do as well.
And then we do also have an additional, an optional way of storing the data in what we
call Thai Flash.
That's a column store.
So by connecting it here you can actually do analytics like aggregations and so on on
the same data within the same transaction even.
And the optimizer here would choose what is the fastest way.
What has the lowest cost for executing the query.
So you don't have any ETL or anything like that in between.
It's very easy to just add.
You're doing all the tables and set the Thai Flash replica equals one or two or if you
add more than one, then you also get the MPP, so massive parallel processing part of it.
We do have an, you can run Spark on it as well.
And let's just go down slightly deeper on how we actually store the data.
So we take all this data and split it into ranges about 100 megabytes and each such range
is stored in three, or yeah it's configurable, let's say three copies in the Thai KV storage
nodes and each such region is forming a raft group.
So that's how it keeps the HA and the high availability.
Thai KV is using ROXDB as lower level storage.
So it's an LSM tree, yeah it's similar as MIROX in Percona or MariaDB.
So it's not B-tree based.
Through this raft protocol, that's how we also can connect the column store.
So that's how we also have it, so you can run it in the same transaction and even if
you have a join, maybe it's faster to execute parts of it through an index in the row store
and then do some of the table scans and aggregation in Thai Flash in the column store.
And this is optional, but this is not, this is the base.
You always need to have the row store and you can have this as an option.
There's a lot of tooling that works.
So first of all, I would say that the data migration, so it's easy to have a ThaiDB cluster
to read the binary logs or just set it up for dumping an upstream MySQL instance or
even several instances into the same cluster so you can combine all the data back.
We have the backup and restore, very good dump story.
I think that even works with MySQL.
You have the tool for do a diff between the different instances, change data capture
that can go to either another ThaiDB cluster or MySQL instance, go through Kafka as well
if you want.
Try up, that's a way for managing and deploying ThaiDB and all components you want.
You can even use it as a playground to start it in your laptop.
It will download the binaries and start everything, including monitoring everything.
So it's very easy to just try out.
We have an operator if you want to run it in Kubernetes as well in the cloud.
So we even have it as a cloud service, you can do anything from on-prem up to a cloud
service in many different ways.
And we also have Lightning, which is an optimized import tool, and that's what I will actually
use in the next slide soon.
A year ago, we started a project because we heard and compared the ad index performance
in ThaiDB cluster versus, for example, Cassandra or Aurora.
And at that time, we were basically three times slower because we haven't optimized
that it was just stable proven and it worked, but it was not fast.
And that's especially when you're doing proof of concept or loading the data, that's where
it's really beneficial to speed it up.
And the way it worked, it would just do data copying through small transaction batches
more or less.
So that also creates a lot of overhead with transaction handling, et cetera.
That's not actually needed when you're doing a backfill because during backfill process,
during the data, it doesn't actually need to be transactional.
And it's only a single node that does this, a single TIDB node that orchestrates it.
I'm not going to go deep into this, it basically just shows how you're creating a command in
one ThaiDB node and it goes into a table, a ThaiDB owner will do it, go through the different
steps and do the data migrations and data copying.
So what we did first was create a feature with this feature flag.
It uses this lightning the import tool technology.
It's completely built in in ThaiDB cluster, so it's not the external tool.
But it reads the data and then it creates these SSD files for ingestion in RocksDB.
So it's very efficient load and it has very low impact on the storage side.
It just moves these files into the storage and enables them and takes them into the
RocksDB levels.
The result was around three times speed up and of course a lot less impact on normal
load in the cluster.
So even if you have a highly loaded cluster, you can do this almost without impacting it.
And then we did a bit of analysis of where we could improve even more and there was things
like the scheduling could be improved just to shorten the time.
Instead of reading directly from the key value store, we could use these co-operators, co-processors
for removing columns that's not needed, for example, for doing optimized scans, etc.
We disconnected the read and write dependencies so they could run in parallel in asynchronous
and a lot of other small optimization.
And that created yet another three to five times speed up.
So all in all, during the last year, we had done 10 times improvement in speed while we're
still only using a single TIDB node and now we're three times faster than the baseline
of the other implementations in Cockroach and Aurora that we have compared with.
And there's a bit more to do, so we're currently looking into how we even can distribute this
instead of running it on a single TIDB node and also being able to auto-tune the priority.
So if you have load that goes a bit up and a bit down, so the DDL work can adjust to
that.
And that is, if you depend on a single TIDB node, if that breaks for any reason, then
your basis is going back to the previous stage, is it?
Yeah, so we have a state state, so we go back a little bit, a little bit, but you don't
need to redo the whole feeling of the index or anything like that.
And yeah, it's all on GitHub.
If you're interested a bit in how it actually works, I would recommend go to OSS Insights.
I would say it's a demo site.
It runs TIDB in the background, and it's a simple web UI, quite nice UI on top.
But it has all of the events from GitHub, so currently it's 5.5 billion records, and
we store it in a single table.
It's a bit other things there as well.
And you can compare for your own GitHub ID or your own project, your own repository,
compare it, and so on, and check some different frameworks, et cetera.
It's quite cool, actually.
Tie-up is very useful if you want to try it on your own laptop or in your own data center.
Of course, you can go to TIDB cloud as well, but I didn't mention that here because that's
our commercial offer.
Something else that we have that is not directly connected, it's chaos mesh.
So if you have a system on Kubernetes and you want to see how it handles different errors,
you can use that for injecting errors.
That's something that we used for stabilizing and testing out the TIDB cluster.
Then I think I'm out of time.
Perfect timing, so you have time to answer questions?
Yeah.
Yeah?
First of all, I'm very interested in how do you organize the htap transitioning.
I mean, you have both storages, and I miss the way you move the data from row into column
or format.
I believe you do double copy.
You have double copies of the data itself.
So we always have the copy here, and the raft leader of the group is always here.
So you do have raft leader and raft follower in the Thai KV, and then we extended the raft
protocol.
So we have learner states here, so they can never become leaders.
So that's how we do.
So this is a must, and this is optional.
What about the optimizer model?
How do you calculate the cost-based approach to understand which storage format you use?
And it's also the influence of the volcano optimizer model, so that's how you more or
less pipeline the different things and can move parts of the pipeline into an MPP framework
that handles the column store.
And I wonder if this model and the optimizer are dispersed across the multiple partitions
of the TIDB operator, or it's in single?
So the optimizer, that's in the TIDB project, in the TIDB repository.
So the SQL node, and when it executes, it's pushed down this co-processor request and
also through the MPP framework for pushing down query fragments or the co-processor request
into either TIDB or a Thai KV or two Thai flash.
So for example, if you're doing a join where one part of the table can be resolved fast
by an index lookup, then it will go here for that part of the table.
And for another table, it might be a big table scan or aggregation that will be faster here.
So then it actually can combine that.
But do you, your cost-based model is based on some assumptions about the cost of these
corporations, right?
I'm not sure.
I don't know the details deep enough for answer that.
And last question, how do you test the compatibility of my SQL client protocol between your implementation
and because it's a big question.
Yeah, yeah.
So we don't have any own connectors or anything like that.
We just relying on MySQL connectors or MariaDB connectors.
And that's what we're using when we're testing.
So you basically have the test use that tries different kind of queries and after they pass
it, you understand that they are somehow equal.
Yeah.
And of course, there are differences.
But I would say the compatibility with the MySQL dialect, it's very, very high.
But of course, like management commands for replication doesn't work because we don't
do replication.
We have internal replication or we use change data capture for transfer to another cluster.
Thank you.
Last question.
What that clash does when there is high rate of single cell updates, like how it handles
this, like rewriting the code files or keeping it separate?
It's a derivative of click house.
So it caches the changes and then it updates it partially or rewrites the whole.
Can this kind of get clicked behind after TKV because it takes more time?
It can.
But if it's behind, then it will more or less fall back here.
You have some tweaking options.
You can even do it as optimizer hints that you want to use either engine, for example,
etc.
Thank you.
If there are more questions, I'm sure Mattias will be able to answer.
Yeah, I'm here.
Even Daniel is here as well.
Thank you.