6 ms·

How does Cockroach efficiently handle the shuffle step when data is on many nodes on the cluster and has to move to be joined? Does Cockroach need high capacit

How does Cockroach efficiently handle the shuffle step when data is on many nodes on the cluster and has to move to be joined? Does Cockroach need high capacity network links to function well? I always see companies making the claim of linear speedup with more nodes but surely that can't be the case if the nodes are geographically disjointed over anything less than gigabit links? Perhaps linear speedup with more nodes is only possible over high speed connections? How high is that exactly? Congratulations to the team on the release! Introducing this kind of database is no easy task - thank you and great job, keep up the good work!

arjunnarayan 9y ago
The short story is we do need high capacity network links to function well. By "high capacity" I mean at least double digit megabit links between your datacenters. A query that inherently requires shuffling because the data is geographically distributed can't get past the bandwidth needs of performing the shuffle. At the very least, with the literal simplest query plan, you're going to need all the raw data to be transported to a single node/datacenter, and I doubt there's a query and network setup where that's more efficient than doing networked shuffles themselves. I don't think you need gigabit networks, but you're certainly going to want at least 10 megabit links. We have not tried to benchmark scenarios where we are bandwidth constrained, so I can't tell you precisely what the minimums are. All the cloud scenarios we've tested (on GCE, Azure, AWS, DigitalOcean) are constrained on other dimensions (i.e. CPU cores, memory, disk IO). And thank you :)
- sixdimensional 9y ago
  That makes sense- I think part of the reason such types of databases are well suited to cloud operations is the guaranteed throughput of the cloud providers own network backbone, which is almost impossible for any single "regular" organization to match, at least for the price. I think we are at a point where doing business without the cloud will become nearly (but not completely) impossible at huge scale with all these features. Thank you very much for your detailed answer and good luck with the continued rollout!
ww520 9y ago
15 years ago I was working on a similar distributed DB product. At the time, the idea was to send the query execution plan to each node to execute any filtering criteria to trim down the candidate row set. Then compute a Bloom Filter on the joining keys on the node with the largest candidate set (using some heuristic statistics), ship the Bloom Filter to other nodes with smaller data set to greatly reduce the non-matching rows. The rows survived the Bloom Filter are highly likely joinable and are shipped back to the main joining node to perform the final join. Bloom Filter is the perfect compromise between size and speed. I'd imagine CockroachDB is doing something similar for distributed join.
- irfansharif 9y ago
  haven't come across this idea before, interesting - will definitely have to give it some more thought. our 'distributed joins', so to speak, run through our distributed query execution model (distsql) setting up incremental 'stages' of computation with the results pipelined and plumbed through individual computes. viewing it through this model our implementation more closely resembles the Grace Hash Join[1] algorithm. you might be interested in the PR[2] that landed this changeset, there's a cool visualization in one of the comments[3] showing the query execution plan. [1]: https://en.wikipedia.org/wiki/Hash_join#Grace_hash_join https://en.wikipedia.org/wiki/Hash_join#Grace_hash_join [2]: https://github.com/cockroachdb/cockroach/pull/12221 https://github.com/cockroachdb/cockroach/pull/12221 [3]: https://github.com/cockroachdb/cockroach/pull/12221#issuecomment-267084941 https://github.com/cockroachdb/cockroach/pull/12221#issuecom...
  - ww520 9y ago
    The Grace Hash Join approach ships the entire joining key set across network. Even if each node just get one partition of it, the aggregate network traffic is the entire set. For small table, it's fine. Large table is going to really tax the network.
- sixdimensional 9y ago
  I've heard of pushdown techniques including function, predicate and aggregate pushdown in distributed relational engines before. Another interesting idea I read about (I can't find it anywhere online) was called "join zippering". Basically you first request the cluster to solve a join by querying and streaming the key columns from a join predicate back into the cluster itself to identify which nodes have matches and then streaming the results from each node in parallel, and doing the join in the stream. This is hard stuff but so cool too :)

arjunnarayan 9y ago

The short story is we do need high capacity network links to function well. By "high capacity" I mean at least double digit megabit links between your datacenters. A query that inherently requires shuffling because the data is geographically distributed can't get past the bandwidth needs of performing the shuffle. At the very least, with the literal simplest query plan, you're going to need all the raw data to be transported to a single node/datacenter, and I doubt there's a query and network setup where that's more efficient than doing networked shuffles themselves. I don't think you need gigabit networks, but you're certainly going to want at least 10 megabit links. We have not tried to benchmark scenarios where we are bandwidth constrained, so I can't tell you precisely what the minimums are. All the cloud scenarios we've tested (on GCE, Azure, AWS, DigitalOcean) are constrained on other dimensions (i.e. CPU cores, memory, disk IO). And thank you :)

sixdimensional 9y ago
That makes sense- I think part of the reason such types of databases are well suited to cloud operations is the guaranteed throughput of the cloud providers own network backbone, which is almost impossible for any single "regular" organization to match, at least for the price. I think we are at a point where doing business without the cloud will become nearly (but not completely) impossible at huge scale with all these features. Thank you very much for your detailed answer and good luck with the continued rollout!

ww520 9y ago

15 years ago I was working on a similar distributed DB product. At the time, the idea was to send the query execution plan to each node to execute any filtering criteria to trim down the candidate row set. Then compute a Bloom Filter on the joining keys on the node with the largest candidate set (using some heuristic statistics), ship the Bloom Filter to other nodes with smaller data set to greatly reduce the non-matching rows. The rows survived the Bloom Filter are highly likely joinable and are shipped back to the main joining node to perform the final join. Bloom Filter is the perfect compromise between size and speed. I'd imagine CockroachDB is doing something similar for distributed join.

irfansharif 9y ago
haven't come across this idea before, interesting - will definitely have to give it some more thought. our 'distributed joins', so to speak, run through our distributed query execution model (distsql) setting up incremental 'stages' of computation with the results pipelined and plumbed through individual computes. viewing it through this model our implementation more closely resembles the Grace Hash Join[1] algorithm. you might be interested in the PR[2] that landed this changeset, there's a cool visualization in one of the comments[3] showing the query execution plan. [1]: https://en.wikipedia.org/wiki/Hash_join#Grace_hash_join https://en.wikipedia.org/wiki/Hash_join#Grace_hash_join [2]: https://github.com/cockroachdb/cockroach/pull/12221 https://github.com/cockroachdb/cockroach/pull/12221 [3]: https://github.com/cockroachdb/cockroach/pull/12221#issuecomment-267084941 https://github.com/cockroachdb/cockroach/pull/12221#issuecom...
- ww520 9y ago
  The Grace Hash Join approach ships the entire joining key set across network. Even if each node just get one partition of it, the aggregate network traffic is the entire set. For small table, it's fine. Large table is going to really tax the network.
sixdimensional 9y ago
I've heard of pushdown techniques including function, predicate and aggregate pushdown in distributed relational engines before. Another interesting idea I read about (I can't find it anywhere online) was called "join zippering". Basically you first request the cluster to solve a join by querying and streaming the key columns from a join predicate back into the cluster itself to identify which nodes have matches and then streaming the results from each node in parallel, and doing the join in the stream. This is hard stuff but so cool too :)