Hello everyone, let's, all right, thank you.
So I'm Harshad Angal, I work for PlanetSkill, and I'm a maintainer of Wittes, and today
I'm going to talk about life of a query.
Why not?
Because I work in query serving team in Wittes.
So what is Wittes basically?
Wittes is nothing but it's used to scale out your MySQL, and it also manages your MySQL
for you.
It's a CNCF gadget project, and it is open source with a partial license 2.0.
So this is the architecture of Wittes, which is basically, there are so many components,
but today we'll just focus on few of them, which is one is VTGate, where your application
connects to, and then we have VT Tablet, where once the application connects to VTGate, then
the queries are sent over there, and then you have the MySQL, where actually your data
are getting stored.
Let's talk about few terminologies, which I'll be using across my presentation, which
is one is key space.
Key space is a logical representation of a database, and it's basically a collection
of your physical databases.
Shard, it's basically a subset of your data, which resides in a key space.
And there's some term called Windex.
Windex is similar to your MySQL indexes, which is basically, it's maintained by the Wittes,
and there's a thing called primary Windexes, which means it will decide where your row
actually live in a particular shard, or in particular table.
So once your query comes in, then it decides, okay, this row, while inserting, where it
should actually, which shard it should actually go to.
So let's take some query, right, then we'll go further.
So let's say we have two tables, customer and order table, and what we want to do is
we want to find a customer who has at least a spend of 1000 bucks basically in their whole
order history.
So we have to take two tables, we take a join, and do a grouping on them, and take a filter
on top of it.
So first thing first.
So the client wants to send a query to you, how they will do it.
So first, the client has to connect to something called VVol VT gates, and so they can use
MySQL protocol.
You have all the MySQL drivers, like in different languages you have, and you can use the already
available MySQL driver to connect directly to VT gate.
VT gate supports the MySQL protocol, so you don't have to do anything on that front.
But it also supports GRPC, and it was supported earlier before we implemented the MySQL protocol,
but it still stays here for its own benefits.
And the reason is like, in MySQL protocol, once you connect, it's like your session is
tied, which means if you open a transaction, you have to commit transaction using that.
But in GRPC, what the initial benefit you get is you can connect to any VT gate, and
you start a transaction over there, and then you can connect to any other VT gate, and
you can commit your transaction using that VT gate.
So let's talk about the different phases that we have to go through.
Once the query is now received by the VT gate, it has to go through parsing, rewriting, planning,
and execution.
And we'll talk in details about each of the phases, but this is what the VT gate does
once it receives the query on its end.
So in the parsing phase, now you receive the query.
Now basically, it will parse it to know whether the query is tactically correct or not.
And once it is tactically correct, then it constructs the abstract syntax tree of it.
So here it will have the select expression from clauses, group by, and the having for
the query we mentioned before.
Now once it's parsed, it goes into the rewriting phase.
And it's very important to have this rewriting phase because we are trying to mimic a single
data store, though your data is distributed across multiple charts.
And this rewriting phase, what we try to do is basically, we first try to see, oh, is
there any literals, is there anything which we can parameterize?
So once we are done with the planning phase, we don't have to plan again and again similar
kind of query.
So what we do is we say, okay, here's IC 1000, but I don't need to plan specifically
for 1000.
So I just make it a parameter, and then I plan for this kind of query rather than planning
with the 1000.
The other thing we do is we do some replacement functions.
So like if you want to get last insert ID, you cannot just send it to any MySQL and get
a last insert ID.
So we do a session management at the VTGate level where for each session, we have to know
what was the last insert ID inserted, so you cannot send directly.
So we need a rewriting phase for these kind of functions.
There are multiple, but I'm just talking about one of them, which is last insert ID.
So you have to do it at the VTGate level.
So then you are replacing that with function with a value, and you know what that value
is for that session.
And the third is you might have to add another SQL node after you construct the AST.
So this is like if on the session you said, I cannot select more than 100 rows.
So after AST, we have in the rewriting phase, we'll add a limit clause as well.
So the planning happens with the limit in place in it.
So after rewriting and the AST generation, before we go into the planning, the planner
needs some additional information, which is it should know what are the key spaces that
exist and what are the shards that the key space map to.
And in the key spaces, what are the tables that exist and what are the indexes that you
use for those tables.
And these information is basically cached in the VTGate, and you have an event watcher
which watches.
And this information comes from, actually, we have something called Topo Server, which
is ZooKeeper at CDSIM, which is where this metadata kind of information is stored.
And let's see what the sharding information looks like that we get from that ZooKeeper.
So what we call is called vschema.
And here it's saying that there's one thing called commerce, and it's sharded.
This is the index that exists.
And these are the two tables, customer and the order table, which use, and they use
the CID column, and they use this index function to shard their table.
So now the VTGate knows about everything, that now it can go into the planning phase.
And the first thing that it does is does the semantic analysis, which is it does basically
scoping, binding, and typing, which is like it's validating, not validated your syntactic
thing before, and now it's trying to validate whether you are using the right columns in
the right places or not, whether they are actually scoped correctly or not, is the visibility
correct or not for those operations, the operations you're using.
And then once it validates that, then it goes into the binding phase, where it binds those
columns to the table that it belongs to, and then it also does the typing, which means
it tries to understand that, okay, once the results come back, we'll know that what type
it will be of.
The second is once the semantic analysis happens, then your AST is converted into an operator
tree, which is basically a logical operator, like you have joints, and then you have tables
and stuff, so it convert into those logical operators.
And then once those logical operator are converted, it then goes into the optimization phase.
So in the optimization phase, we basically make a decision that how the plan should look
like, and in the innovators, basically what we do is we get the SQL, and ultimately after
the planning phase, we'll get another SQL, because the data does not reside on the VTGit,
it resides on the MySQL, and we want to optimize on the way that we should get less and less
information back on the VTGit to process your data.
So we try to push as much as down to the query to the MySQL so that that can resolve much
faster for us than we do it at the VTGit.
So in the optimization phase, basically after the optimization completes, we call something
like routes, which tell us that, okay, this query will go to which all shards, so that's
what the ultimate step becomes, and that's where we call it as a physical operator.
And then once we have the physical operators, we transform the plan into an execution plan.
And the execution plan is nothing but a collection of engine primitives that we call.
So let's look at how the execution plan looks like.
So if both the tables which I showed in the V schema were using same sharding function,
so that's why we were able to basically make it as a single route, which is telling you
that, okay, just go scatter to all the shards and just gather the result at the VTGit level
and send it back to the user.
But if both the tables, the customer table and the order table were using some different
sharding functions, then you cannot merge it like that.
Then it says basically, it will look something like that, like you'll have, now, because
it gets very complex enough to show everything over here, so I'm just showing that there's
a, there'll be two routes that will happen, and then ultimately the join will also happen
at the VTGit level, sorting and projecting and the aggregation, and then finally the
filtering based on how much customers have spent those thousand bucks.
So everything happens on the VTGit.
Let's just look at the two routes that I showed on the left.
So first one we'll do is basically doing a scatter, and it's trying to do on the order
table that, okay, give me for all the customer IDs what is the spend they have done, and
then on the right-hand side, we need the customer emails.
So for every customer ID, I need to get their email IDs, and because the customer ID was
the sharding key, the primary index for that table, we are able to send the second query
as a, only to a specific shard, we don't have to do again, go to all the shards.
So what all steps happened in the execution?
So first thing is, once we get the route, we resolve it using those indexes, and so
that it can go to the specific correct shard.
Another is then, it basically then we take a decide, okay, now we know which shard we
have to go to, then it will also decide, okay, this is a query that we have to send it to
something that we call the VT tablets, which basically in the shard we have VT tablets
and MySQL, so it's sent to the VT tablet, and then once you receive the result back,
you gather the information, but sometimes you also have to do transactions, so if transaction
is needed, then the VT gate will also tell VT tablets, go and also open the transaction
when you execute the query, and then the session, the transactions will be managed by the VT
gate that will know in which shard what is the transaction being opened, so that when
you do the commit, it knows which shard it has to send the commit information as well
too.
And the last thing that execution also handles is basically, if you have a select query and
you want to read it on the replicas, then it also does the load balancing for you, so
it doesn't overload a single replica.
So now the query is being sent to VT tablets, and now the VT job is to basically get the
result back to the VT gate.
So what are the things that goes in the VT tablet?
It's exactly similar thing that goes in the VT tablets as well, but what are different
reasons?
So first thing is, VT gate has some view of the VT tablet of what this state is, but the
first thing that events the query received by the VT tablet, it does this, okay, let
me validate whether that's current, am I able to serve this query or not, if any view
has changed before, after sending the query and before it reads the VT tablet, so it validates
if everything is okay, okay, then let's go and do the parsing.
So it will parse the query, and the reason is it has to parse the query is because VT
tablet has its checks and balances over here where it tries to put some limitation, like
it says that don't overload the MySQL resource, so it will basically add its own limits onto
the query so that it doesn't overload your MySQL, that you can configure and do things,
and then it goes into the planning phase, and in the planning field, it basically tries
to see whether did you have put any query rules, which means like any bad query, if
you put like, okay, somebody does select star and without a wear condition and stuff, so
it basically says, okay, this query cannot be executed because it's a bad query, so
those kind of query rules you can put, and another thing that we check in the planning
phase is basically it will say, okay, this is the user that sent this query, but is the
user allowed to even access the tables or not?
So if it's not allowed, then we throw the error back.
After the planning phase, also VT tablet does this, again, and all these are there to just
not to load your MySQL, so another thing that it does is query consolidation, which is basically
it checks whether any same query is running on the MySQL or not.
If it's already running, then it just waits for that query to return the result, and all
the threads which are waiting for the same query will get the result and it will return
back, so only one query ultimately gets executed on the MySQL and not all the queries get executed
with the same exact same query, yes.
So once it thinks, okay, nothing can be done, like it has to finally send it to MySQL, then
it will use one of the connection from the connection pool to send the query to MySQL,
and once the results are written back, it will send it to VTGate, VTGate, if it has to
do some operations, it will do it, and finally the client will receive the query.
So yeah, so that's what the life of query is, but then there are some custom operations
which also affect your query path, and which is like when there's a plan maintenance going,
like you're promoting one replica to primary for some reason, and there is basically if
you are splitting your data, like you have some N charts and you want to go to two N
charts or N plus one charts or such thing like that, so while doing those operations,
what VTGate does is it notifies the topo that some operation is going, VTGate understands
it, so it doesn't send query to the VTGate tablet, it basically buffers a certain duration,
and once everything comes back right, then the queries start going to the VTGate tablet,
or it times out and VTGate does the time out from itself rather than sending it down.
So yeah, thank you.
Questions?
So currently we don't do cross-shared transactions, we do it, but in the best effort way, but
it's currently on the application to know that they are doing cross-shared transactions,
but we allow it, but the application should know that they are going cross-shared.
Yes?
So join I already showed in how we have the two routes, and then from one we get the
result and the other one, this is join.
So we have hash joins available, but then it will consume your memory, and otherwise
it's like from one you'll get the result and then you send it to the other one, so
it's just sequential.
Aggregation happens at the VTGate level, he has a sorting layer, so before aggregation
we have to sort, yes, that's what it was there in the diagram as well, that after joint
there was sort layer, because you have to sort, so we have an in-memory sort, so there
are multiple again sort based on what we can do the best, so there's a merge sort also
and then you have a complete sort as well, based on what can we do best.
And then sorting, then it goes to the aggregation.
So that's why we said we try to push as much as with MySQL, so we actually push the sort
as well to MySQL if possible for the merge sort, but if you do it in the memory then
we still sort it, so it depends whether we can push it down or not, if you are able to
push it down we push as much as to the MySQL, yes, thank you, all right I think we have
a question, thank you very much, thank you.
You