ignite-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Sergi Vladykin <sergi.vlady...@gmail.com>
Subject Distributive SQL Joins
Date Fri, 07 Aug 2015 06:29:29 GMT

I'm going to start working on distributed SQL joins soon and want to put my
thoughts on this matter here.

To provide collocated and non-collocated joins we need to know affinity key
for each table.
For example we have a table `Organization(id int)` where `id` is affinity
key and `Person(orgId int)`
where `orgId` is affinity key. This way when we see join `Person p,
Organization o on p.orgId = o.id`
we know that it does not make sense to try to find joins between different
nodes because
if the affinity fields are equal then they must be on the same node. Also
this allows to handle
transitive collocated joins like `Person pe, Product pr join on pe.orgId =
pr.orgId` where `orgId`
is affinity key for `Product` but is not a primary key. This will be a
backward incompatible configuration change.
This configuration approach is consistent with what MemSQL does.

Obviously we can have multiple tables joined on different keys in the same
Lets say `+` is a join on collocated key and `-` is a join on
non-collocated key.
Suppose we have the following join in query
`a + b + c - d + e - f`

A. We can run it the following way with shuffle:
run collocated `(a + b + c) = m` , `(d + e) = n` resulting into
`m - n - f`
Then there are two possibilities (because we have only 3 non-collocated
either they joined on the same key or not. In the the first case we can
shuffle them
in a single step `(m - n - f) = z` to achieve collocation on joined fields,
in the second we have
to do shuffle `(m - n) = k` and then shuffle `(k - f) = z` which will be
resulting entity which is known to be collocated on joined fields of `k`
and `f`.

B. Another approach is to request data from remote nodes as needed for a
It means that we are running locally `(a + b + c)` and when we fetch a row
from there, we request a joining part for this row from `(d + e)` part.
And if it is known that we join on affinity key from `(d + e)` then we know
on which node this part exists exactly. Otherwise we have to broadcast this
After that the same must happen with `f` part. Of course batching here is
as well, no need to request data for each node separately.

If we analyze these two approaches from the performance standpoint then
we can see that the best one is B with known affinity key of remote side.
It has the same number of messages as shuffle but reduces traffic because
on request we need to send only keys (while on shuffle we need to send the
whole local table part to be joined on a third node) and in response we
will receive only
joined row parts (while in shuffle we need to send the whole remote
table part to be joined on a third node).

Obviously B without known affinity key requires broadcast and is not
useful. But it seems that such a case will be quite uncommon since it is a
join of two partitioned tables which are in turn collocated with two
different partitioned tables.

Since implementing approach A is much more complex (it will require more
query planning and generation as well as more complex coordination between
and B with known affinity keys is simpler and more effective
my preference is to implement B with known affinity keys, and forbid
case when we don't know remote affinity key.

Feel free to comment.


  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message