{"action":"created","issue":{"url":"https://api.github.com/repos/basho/riak_kv/issues/1538","repository_url":"https://api.github.com/repos/basho/riak_kv","labels_url":"https://api.github.com/repos/basho/riak_kv/issues/1538/labels{/name}","comments_url":"https://api.github.com/repos/basho/riak_kv/issues/1538/comments","events_url":"https://api.github.com/repos/basho/riak_kv/issues/1538/events","html_url":"https://github.com/basho/riak_kv/pull/1538","id":189464868,"number":1538,"title":"Parallel decode of query results","user":{"login":"erikleitch","id":9666459,"avatar_url":"https://avatars.githubusercontent.com/u/9666459?v=3","gravatar_id":"","url":"https://api.github.com/users/erikleitch","html_url":"https://github.com/erikleitch","followers_url":"https://api.github.com/users/erikleitch/followers","following_url":"https://api.github.com/users/erikleitch/following{/other_user}","gists_url":"https://api.github.com/users/erikleitch/gists{/gist_id}","starred_url":"https://api.github.com/users/erikleitch/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/erikleitch/subscriptions","organizations_url":"https://api.github.com/users/erikleitch/orgs","repos_url":"https://api.github.com/users/erikleitch/repos","events_url":"https://api.github.com/users/erikleitch/events{/privacy}","received_events_url":"https://api.github.com/users/erikleitch/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":4,"created_at":"2016-11-15T18:07:40Z","updated_at":"2016-11-30T16:43:48Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/basho/riak_kv/pulls/1538","html_url":"https://github.com/basho/riak_kv/pull/1538","diff_url":"https://github.com/basho/riak_kv/pull/1538.diff","patch_url":"https://github.com/basho/riak_kv/pull/1538.patch"},"body":"<hr>\r\n\r\n**Summary**\r\n\r\nThis PR implements changes to decode SQL SELECT query results at the vnodes.  Currently, SQL queries are split into subqueries by the coordinating node, each vnode in the coverage plan executes the subquery as a fold over keys in eleveldb, and the results are returned to the coordinating vnode, which then decodes the results from on-disk format to erlang terms.\r\n\r\nThe problem with this architecture is that although large queries can be parallelized across the cluster by tailoring TS quanta to split large queries into many subqueries that execute as parallel folds, the decoding is still serialized at the coordinating node.  For large queries, the decoding dominates the query latency, so we therefore lose the benefit of parallelizing the queries in the first place.\r\n\r\nThis change delegates decoding of the query results to the vnodes in the coverage plan, so that both the eleveldb fold and the decoding of the resulting data can be effectively parallelized across the cluster.  This results in a > N-fold speedup for large queries, where N is the number of physical nodes in the cluster\r\n\r\n<hr>\r\n\r\n**Overall Design**\r\n\r\nDelegating decoding to the vnodes instead of the coordinator is straightforward, but to ensure functionality in mixed clusters, the implementation is a bit more complicated.  In summary, the design is:\r\n\r\n- we register a new capability `{riak_kv, decode_query_results_at_vnode}` for riak_kv, indicating whether or not decoding can be done at the vnode.\r\n\r\n- the coordinator sends query requests to the vnodes as before -- i.e., no modifications have been made to the request path.  This ensures that query requests can be sent from either an old or new coordinator to either an old or new vnode.\r\n\r\n- the new vnode checks `{riak_kv, decode_query_results_at_vnode}` to determine whether results should be sent decoded or encoded. If true, the vnode sends a decoded Chunk, as `{results, {decoded, Chunk}}`. If false, the vnode sends an encoded Chunk, as `{results, Chunk}` (current behavior)\r\n\r\n  This allows the coordinator to handle either decoded or encoded data, by pattern-matching on the message structure.\r\n\r\n<hr>\r\n\r\n**Details**\r\n\r\n`src/riak_kv_app.erl` \r\n\r\n- Modified to register the new `{riak_kv, decode_query_results_at_vnode}` capability.\r\n\r\n`src/riak_kv_qry_worker.erl` \r\n    \r\n- Now exports `decode_results/1` because it may be used either at the vnode (in `riak_kv_vnode.erl`) or here by the coordinator.  \r\n- Helper function `get_decoded_results/1` has been added to handle either encoded or decoded data, depending on whether the sending vnode is old or new.  \r\n\r\n`src/riak_kv_vnode.erl` \r\n    \r\n- A range-scan-specific result acknowledgement function `range_scan_result_fun_ack/2` replaces the default ack function for query requests.  \r\n- `range_scan_result_fun_ack/2` checks cluster capabilities, and if `{riak_kv, decode_query_results_at_vnode}` capability is `true`, decodes the results before sending back to the coordinator (if `false` it sends back encoded results as before)\r\n\r\n<hr>\r\n\r\n**Backwards/Downgrade Compatibility**\r\n\r\nThe aim of the design is to support functionality in mixed clusters, where we may have _either_ old or new coordinators talking to _either_ old or new vnodes; in short, there should be no impact on functionality if a customer upgrades to TS1.5, and subsequently downgrades some nodes.  I summarize the interaction for each of the 4 possible scenarios below:\r\n\r\n1. new coordinator + new vnode\r\n\r\n   - new vnode checks capabilities, sends back _decoded_\r\n   data, since `{riak_kv, decode_query_results_at_vnode} == true`\r\n   - new vnode sends back decoded data as `{results, {decoded, Chunk}}`\r\n   - new coordinator knows not to decode the result because of pattern match on `{results, {decoded, Chunk}}`\r\n\r\n2. old coordinator + new vnode\r\n\r\n   - new node checks capabilities, sends back _encoded_\r\n   data, since `{riak_kv, decode_query_results_at_vnode} == false`\r\n   - new vnode sends back encoded data as `{results, Chunk}`, as before\r\n   - old coordinator decodes the data as before\r\n\r\n3. new coordinator + old vnode\r\n\r\n   - old vnode sends back encoded data, as before\r\n   - old vnode sends back encoded data as `{results, Chunk}`\r\n   - new coordinator knows to decode the result because of pattern match on `{results, Chunk}`\r\n\r\n4. old coordinator + old vnode\r\n\r\n   - as before\r\n\r\n<hr>\r\n\r\n**Performance**\r\n\r\nPerformance with this branch is described in detail here: https://github.com/basho/internal_wiki/wiki/Performance-Improvements-for-Riak#ts1.5.  \r\n\r\n_Disclaimer: the following discussion presents results from TS1.4 with the modifications in this PR, except for the capabilities lookup.  I'm doing this 1. because I already have those results handy, and I'm still working on the comparison with this branch in TS1.5-alpha, and 2. because if we remove the TTB encoding as planned, these results are more representative of the performance boost we should expect._  \r\n\r\nI reproduce two relevant plots below.  Each plot shows query latencies for a range of table schema sizes and query sizes, for queries executed monolithically against a single partition (left panels),  the same queries split into sub-queries across 3 nodes (middle panels) and the ratio of those latencies (right panels).\r\n\r\nThe first demonstrates that the change behaves as expected.  Namely, that by reducing the quantum so that large queries are parallelized across the cluster, the ratio approaches ~N for large queries (in this case N = 3):\r\n\r\n![tsquerylatency_qcmp_prototype](https://cloud.githubusercontent.com/assets/9666459/20320959/99cfca60-ab28-11e6-913f-c3cdcb6855b6.png)\r\n\r\nThe second compares monolithic queries between TS1.4 and this branch, demonstrating that the additional overhead from disterl handling of decoded terms has a negligible impact on the query latency (ratio is consistent with 1):\r\n\r\n![tsquerylatency_qcmp_single](https://cloud.githubusercontent.com/assets/9666459/20326519/1277ed3e-ab3e-11e6-91d3-ec8b3291111c.png)\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/basho/riak_kv/issues/comments/263925142","html_url":"https://github.com/basho/riak_kv/pull/1538#issuecomment-263925142","issue_url":"https://api.github.com/repos/basho/riak_kv/issues/1538","id":263925142,"user":{"login":"paegun","id":824570,"avatar_url":"https://avatars.githubusercontent.com/u/824570?v=3","gravatar_id":"","url":"https://api.github.com/users/paegun","html_url":"https://github.com/paegun","followers_url":"https://api.github.com/users/paegun/followers","following_url":"https://api.github.com/users/paegun/following{/other_user}","gists_url":"https://api.github.com/users/paegun/gists{/gist_id}","starred_url":"https://api.github.com/users/paegun/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/paegun/subscriptions","organizations_url":"https://api.github.com/users/paegun/orgs","repos_url":"https://api.github.com/users/paegun/repos","events_url":"https://api.github.com/users/paegun/events{/privacy}","received_events_url":"https://api.github.com/users/paegun/received_events","type":"User","site_admin":false},"created_at":"2016-11-30T16:43:48Z","updated_at":"2016-11-30T16:43:48Z","body":"+1 b1970a2\r\n\r\nI agree re: `lists:flatten` suggestion, but also agree that the local performance cost is far outweighed by the overarching performance improvement."}}
{"actor":{"display_login":"paegun"}}