{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/5725","labels_url":"https://api.github.com/repos/apache/spark/issues/5725/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/5725/comments","events_url":"https://api.github.com/repos/apache/spark/issues/5725/events","html_url":"https://github.com/apache/spark/pull/5725","id":71366914,"number":5725,"title":"[SPARK-7076][SPARK-7077][SPARK-7080][SQL] Use managed memory for aggregations","user":{"login":"JoshRosen","id":50748,"avatar_url":"https://avatars.githubusercontent.com/u/50748?v=3","gravatar_id":"","url":"https://api.github.com/users/JoshRosen","html_url":"https://github.com/JoshRosen","followers_url":"https://api.github.com/users/JoshRosen/followers","following_url":"https://api.github.com/users/JoshRosen/following{/other_user}","gists_url":"https://api.github.com/users/JoshRosen/gists{/gist_id}","starred_url":"https://api.github.com/users/JoshRosen/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/JoshRosen/subscriptions","organizations_url":"https://api.github.com/users/JoshRosen/orgs","repos_url":"https://api.github.com/users/JoshRosen/repos","events_url":"https://api.github.com/users/JoshRosen/events{/privacy}","received_events_url":"https://api.github.com/users/JoshRosen/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":16,"created_at":"2015-04-27T18:36:58Z","updated_at":"2015-04-28T20:52:18Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/5725","html_url":"https://github.com/apache/spark/pull/5725","diff_url":"https://github.com/apache/spark/pull/5725.diff","patch_url":"https://github.com/apache/spark/pull/5725.patch"},"body":"This patch adds managed-memory-based aggregation to Spark SQL / DataFrames. Instead of working with Java objects, this new aggregation path uses `sun.misc.Unsafe` to manipulate raw memory.  This reduces the memory footprint for aggregations, resulting in fewer spills, OutOfMemoryErrors, and garbage collection pauses.  As a result, this allows for higher memory utilization.  It can also result in better cache locality since objects will be stored closer together in memory.\r\n\r\nThis feature can be eanbled by setting `spark.sql.unsafe.enabled=true`.  For now, this feature is only supported when codegen is enabled and only supports aggregations for which the grouping columns are primitive numeric types or strings and aggregated values are numeric.\r\n\r\n### Managing memory with sun.misc.Unsafe\r\n\r\nThis patch supports both on- and off-heap managed memory.\r\n\r\n- In on-heap mode, memory addresses are identified by the combination of a base Object and an offset within that object.\r\n- In off-heap mode, memory is addressed directly with 64-bit long addresses.\r\n\r\nTo support both modes, functions that manipulate memory accept both `baseObject` and `baseOffset` fields.  In off-heap mode, we simply pass `null` as `baseObject`.\r\n\r\nWe allocate memory in large chunks, so memory fragmentation and allocation speed are not significant bottlenecks.\r\n\r\nBy default, we use on-heap mode.  To enable off-heap mode, set `spark.unsafe.offHeap=true`.\r\n\r\n### Compact tuple format\r\n\r\nThis patch introduces `UnsafeRow`, a compact row layout.  In this format, each tuple has three parts: a null bit set, fixed length values, and variable-length values:\r\n\r\n![image](https://cloud.githubusercontent.com/assets/50748/7328538/2fdb65ce-ea8b-11e4-9743-6c0f02bb7d1f.png)\r\n\r\n- Rows are always 8-byte word aligned (so their sizes will always be a multiple of 8 bytes)\r\n- The bit set is used for null tracking:\r\n\t- Position _i_ is set if and only if field _i_ is null\r\n\t- The bit set is aligned to an 8-byte word boundary.\r\n- Every field appears as an 8-byte word in the fixed-length values part:\r\n\t- If a field is null, we zero out the values.\r\n\t- If a field is variable-length, the word stores a relative offset (w.r.t. the base of the tuple) that points to the beginning of the field's data in the variable-length part.\r\n- Each variable-length data type can have its own encoding:\r\n\t- For strings, the first word stores the length of the string and is followed by UTF-8 encoded bytes.  If necessary, the end of the string is padded with empty bytes in order to ensure word-alignment.\r\n\r\nFor example, a tuple that consists 3 fields of type (int, string, string), with value (null, “data”, “bricks”) would look like this:\r\n\r\n![image](https://cloud.githubusercontent.com/assets/50748/7328526/1e21959c-ea8b-11e4-9a28-a4350fe4a7b5.png)\r\n\r\nThis format allows us to compare tuples for equality by directly comparing their raw bytes.  This also enables fast hashing of tuples.\r\n\r\n### Hash map for performing aggregations\r\n\r\nThis patch introduces `UnsafeFixedWidthAggregationMap`, a hash map for performing aggregations where the aggregation result columns are fixed-with.  This map's keys and values are `Row` objects. `UnsafeFixedWidthAggregationMap` is implemented on top of `BytesToBytesMap`, an append-only map which supports byte-array keys and values.\r\n\r\n`BytesToBytesMap` stores pointers to key and value tuples.  For each record with a new key, we copy the key and create the aggregation value buffer for that key and put them in a buffer. The hash table then simply stores pointers to the key and value. For each record with an existing key, we simply run the aggregation function to update the values in place.\r\n\r\nThis map is implemented using open hashing with triangular sequence probing.  Each entry stores two words in a long array: the first word stores the address of the key and the second word stores the relative offset from the key tuple to the value tuple, as well as the key's 32-bit hashcode.  By storing the full hashcode, we reduce the number of equality checks that need to be performed to handle position collisions ()since the chance of hashcode collision is much lower than position collision).\r\n\r\n`UnsafeFixedWidthAggregationMap` allows regular Spark SQL `Row` objects to be used when probing the map.  Internally, it encodes these rows into `UnsafeRow` format using `UnsafeRowConverter`.  This conversion has a small overhead that can be eliminated in the future once we use UnsafeRows in other operators.\r\n\r\n<!-- Reviewable:start -->\r\n[<img src=\"https://reviewable.io/review_button.png\" height=40 alt=\"Review on Reviewable\"/>](https://reviewable.io/reviews/apache/spark/5725)\r\n<!-- Reviewable:end -->\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/97203679","html_url":"https://github.com/apache/spark/pull/5725#issuecomment-97203679","issue_url":"https://api.github.com/repos/apache/spark/issues/5725","id":97203679,"user":{"login":"AmplabJenkins","id":4031291,"avatar_url":"https://avatars.githubusercontent.com/u/4031291?v=3","gravatar_id":"","url":"https://api.github.com/users/AmplabJenkins","html_url":"https://github.com/AmplabJenkins","followers_url":"https://api.github.com/users/AmplabJenkins/followers","following_url":"https://api.github.com/users/AmplabJenkins/following{/other_user}","gists_url":"https://api.github.com/users/AmplabJenkins/gists{/gist_id}","starred_url":"https://api.github.com/users/AmplabJenkins/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/AmplabJenkins/subscriptions","organizations_url":"https://api.github.com/users/AmplabJenkins/orgs","repos_url":"https://api.github.com/users/AmplabJenkins/repos","events_url":"https://api.github.com/users/AmplabJenkins/events{/privacy}","received_events_url":"https://api.github.com/users/AmplabJenkins/received_events","type":"User","site_admin":false},"created_at":"2015-04-28T20:52:18Z","updated_at":"2015-04-28T20:52:18Z","body":"Merged build started."}}