GitHub PullRequestReviewCommentEvent at 2014-04-08 06:23:17
Repository: presto
Actor: martint
Payload: {"comment":{"url":"https://api.github.com/repos/facebook/presto/pulls/comments/11377694","id":11377694,"diff_hunk":"@@ -89,156 +88,173 @@ public long getEstimatedSize()\n         return types;\n     }\n \n+    public int getGroupCount()\n+    {\n+        return nextGroupId;\n+    }\n+\n+    public void appendValuesTo(int groupId, BlockBuilder[] builders)\n+    {\n+        long address = groupAddress.get(groupId);\n+        PageBuilder page = pages.get(decodeSliceIndex(address));\n+        page.appendValuesTo(decodePosition(address), builders);\n+    }\n+\n     public GroupByIdBlock getGroupIds(Page page)\n     {\n         int positionCount = page.getPositionCount();\n \n-        int groupIdBlockSize = SINGLE_LONG.getFixedSize() * positionCount;\n-        BlockBuilder blockBuilder = new BlockBuilder(SINGLE_LONG, groupIdBlockSize, Slices.allocate(groupIdBlockSize).getOutput());\n+        int maxPossibleGroupId = nextGroupId + positionCount;\n+        groupAddress.ensureCapacity(maxPossibleGroupId);\n+        if (maxPossibleGroupId > maxFill) {\n+            rehash(maxPossibleGroupId);\n+        }\n+\n+        // we know the exact size required for the block\n+        BlockBuilder blockBuilder = BIGINT.createFixedSizeBlockBuilder(positionCount);\n \n         // open cursors for group blocks\n-        BlockCursor[] cursors = new BlockCursor[channels.length];\n+        BlockCursor[] currentRow = new BlockCursor[channels.length];\n         for (int i = 0; i < channels.length; i++) {\n-            cursors[i] = page.getBlock(channels[i]).cursor();\n+            currentRow[i] = page.getBlock(channels[i]).cursor();\n         }\n \n-        // use cursors in hash strategy to provide value for \"current\" row\n-        hashStrategy.setCurrentRow(cursors);\n-\n-        for (int position = 0; position < positionCount; position++) {\n-            for (BlockCursor cursor : cursors) {\n+        // index pages\n+        for (int position = 0; position < page.getPositionCount(); position++) {\n+            for (BlockCursor cursor : currentRow) {\n                 checkState(cursor.advanceNextPosition());\n             }\n \n-            int groupId = pagePositionToGroupId.get(CURRENT_ROW_ADDRESS);\n-            if (groupId < 0) {\n-                groupId = addNewGroup(cursors);\n-            }\n+            // get the group for the current row\n+            int groupId = putIfAbsent(currentRow);\n+\n+            // output the group id for this row\n             blockBuilder.append(groupId);\n         }\n-        UncompressedBlock block = blockBuilder.build();\n+\n+        RandomAccessBlock block = blockBuilder.build();\n         return new GroupByIdBlock(nextGroupId, block);\n     }\n \n-    private int addNewGroup(BlockCursor... row)\n+    private int putIfAbsent(BlockCursor[] cursors)\n     {\n-        int pageIndex = allPages.size() - 1;\n-        if (!activePage.append(row)) {\n-            // record the active page memory size\n-            completedPagesMemorySize += activePage.getMemorySize();\n+        int hashPosition = (murmurHash3(hashCursor(cursors) ^ mask)) & mask;\n \n-            activePage = new GroupByPageBuilder(types);\n-            allPages.add(activePage);\n-            pageIndex++;\n+        // look for an empty slot or a slot containing this key\n+        int groupId = -1;\n+        while (key[hashPosition] != -1) {\n+            long address = key[hashPosition];\n+            if (positionEqualsCurrentRow(decodeSliceIndex(address), decodePosition(address), cursors)) {\n+                // found an existing slot for this key\n+                groupId = value[hashPosition];\n \n-            // TODO make the page builder allocation guarantee enough space to hold at least the first row.\n-            checkState(activePage.append(row), \"Could not add row to empty page builder\");\n+                break;\n+            }\n+            // increment position and mask to handler wrap around\n+            hashPosition = (hashPosition + 1) & mask;\n         }\n \n-        // record group id in hash\n-        int groupId = nextGroupId++;\n-        long address = encodeSyntheticAddress(pageIndex, activePage.getPositionCount() - 1);\n-        pagePositionToGroupId.put(address, groupId);\n-\n+        // did we find an existing group?\n+        if (groupId < 0) {\n+            groupId = addNewGroup(hashPosition, cursors);\n+        }\n         return groupId;\n     }\n \n-    public Long2IntOpenCustomHashMap getPagePositionToGroupId()\n-    {\n-        return pagePositionToGroupId;\n-    }\n-\n-    public void appendValuesTo(long pagePosition, BlockBuilder[] builders)\n-    {\n-        GroupByPageBuilder page = allPages.get(decodeSliceIndex(pagePosition));\n-        page.appendValuesTo(decodePosition(pagePosition), builders);\n-    }\n-\n-    private class PageBuilderHashStrategy\n-            implements Strategy\n+    private int addNewGroup(int hashPosition, BlockCursor[] cursors)\n     {\n-        private BlockCursor[] currentRow;\n+        // add the row to the open page\n+        int pageIndex = pages.size() - 1;\n+        PageBuilder pageBuilder = pages.get(pageIndex);\n+        pageBuilder.append(cursors);\n \n-        public void setCurrentRow(BlockCursor[] currentRow)\n-        {\n-            this.currentRow = currentRow;\n-        }\n+        // record group id in hash\n+        int groupId = nextGroupId++;\n+        long address = encodeSyntheticAddress(pageIndex, pageBuilder.getPositionCount() - 1);\n \n-        @Override\n-        public int hashCode(long sliceAddress)\n-        {\n-            if (sliceAddress == CURRENT_ROW_ADDRESS) {\n-                return hashCurrentRow();\n-            }\n-            else {\n-                return hashPosition(sliceAddress);\n-            }\n-        }\n+        key[hashPosition] = address;\n+        value[hashPosition] = groupId;\n+        groupAddress.set(groupId, address);\n \n-        private int hashPosition(long sliceAddress)\n-        {\n-            int sliceIndex = decodeSliceIndex(sliceAddress);\n-            int position = decodePosition(sliceAddress);\n-            return allPages.get(sliceIndex).hashCode(position);\n-        }\n+        // create new page builder if this page is full\n+        if (pageBuilder.isFull()) {\n+            completedPagesMemorySize += pageBuilder.getMemorySize();\n \n-        private int hashCurrentRow()\n-        {\n-            int result = 0;\n-            for (int channel = 0; channel < types.size(); channel++) {\n-                Type type = types.get(channel);\n-                BlockCursor cursor = currentRow[channel];\n-                result = addToHashCode(result, valueHashCode(type, cursor.getRawSlice(), cursor.getRawOffset()));\n-            }\n-            return result;\n+            pageBuilder = new PageBuilder(types);\n+            pages.add(pageBuilder);\n         }\n+        return groupId;\n+    }\n \n-        @Override\n-        public boolean equals(long leftSliceAddress, long rightSliceAddress)\n-        {\n-            // current row always equals itself\n-            if (leftSliceAddress == CURRENT_ROW_ADDRESS && rightSliceAddress == CURRENT_ROW_ADDRESS) {\n-                return true;\n+    private void rehash(int size)\n+    {\n+        int newSize = arraySize(size + 1, FILL_RATIO);\n+\n+        int newMask = newSize - 1; // Note that this is used by the hashing macro","path":"presto-main/src/main/java/com/facebook/presto/operator/GroupByHash.java","position":288,"original_position":288,"commit_id":"15e3683e24b5240b44b9234308375280c4189f3b","original_commit_id":"15e3683e24b5240b44b9234308375280c4189f3b","user":{"login":"martint","id":14387,"avatar_url":"https://avatars.githubusercontent.com/u/14387?","gravatar_id":"3cca33f9fc2e317da7e2b69c7a05c4ab","url":"https://api.github.com/users/martint","html_url":"https://github.com/martint","followers_url":"https://api.github.com/users/martint/followers","following_url":"https://api.github.com/users/martint/following{/other_user}","gists_url":"https://api.github.com/users/martint/gists{/gist_id}","starred_url":"https://api.github.com/users/martint/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/martint/subscriptions","organizations_url":"https://api.github.com/users/martint/orgs","repos_url":"https://api.github.com/users/martint/repos","events_url":"https://api.github.com/users/martint/events{/privacy}","received_events_url":"https://api.github.com/users/martint/received_events","type":"User","site_admin":false},"body":"What's the \"hashing macro\"?","created_at":"2014-04-08T06:23:17Z","updated_at":"2014-04-08T06:23:17Z","html_url":"https://github.com/facebook/presto/pull/981#discussion_r11377694","pull_request_url":"https://api.github.com/repos/facebook/presto/pulls/981","_links":{"self":{"href":"https://api.github.com/repos/facebook/presto/pulls/comments/11377694"},"html":{"href":"https://github.com/facebook/presto/pull/981#discussion_r11377694"},"pull_request":{"href":"https://api.github.com/repos/facebook/presto/pulls/981"}}}}