IGNITE-3477 - Fixing query parallelism

apache · Mar 22, 2017 · aaf23fd · aaf23fd
1 parent 6cc1403
commit aaf23fd
Show file tree

Hide file tree

Showing 5 changed files with 80 additions and 36 deletions.
diff --git a/...es/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java b/...es/core/src/main/java/org/apache/ignite/internal/processors/query/GridQueryProcessor.java
@@ -1761,7 +1761,7 @@ private static ClassProperty buildClassProperty(Class<?> keyCls, Class<?> valCls
      * @param valCls value class
      * @param pathStr property name
      * @param resType property type
-     * @return
+     * @return Property initialization exception.
      */
     public static String propertyInitializationExceptionMessage(Class<?> keyCls, Class<?> valCls, String pathStr,
         Class<?> resType) {

diff --git a/.../core/src/test/java/org/apache/ignite/internal/processors/database/BPlusTreeSelfTest.java b/.../core/src/test/java/org/apache/ignite/internal/processors/database/BPlusTreeSelfTest.java
@@ -1203,7 +1203,7 @@ private void doTestRandomPutRemoveMultithreaded(boolean canGetRow) throws Except
 
         final Map<Long,Long> map = new ConcurrentHashMap8<>();
 
-        final int loops = reuseList == null ? 200_000 : 400_000;
+        final int loops = reuseList == null ? 100_000 : 200_000;
 
         final GridStripedLock lock = new GridStripedLock(256);
 
@@ -1295,7 +1295,7 @@ else if (op == 3) {
 
                 return null;
             }
-        }, 16, "put-remove");
+        }, Runtime.getRuntime().availableProcessors(), "put-remove");
 
         final AtomicBoolean stop = new AtomicBoolean();
 

diff --git a/...dexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java b/...dexing/src/main/java/org/apache/ignite/internal/processors/query/h2/IgniteH2Indexing.java
@@ -2978,7 +2978,9 @@ private Index createSortedIndex(
                 if (log.isInfoEnabled())
                     log.info("Creating cache index [cacheId=" + cctx.cacheId() + ", idxName=" + name + ']');
 
-                return new H2TreeIndex(cctx, tbl, name, pk, cols, inlineSize);
+                final int segments = tbl.rowDescriptor().configuration().getQueryParallelism();
+
+                return new H2TreeIndex(cctx, tbl, name, pk, cols, inlineSize, segments);
             }
             catch (IgniteCheckedException e) {
                 throw new IgniteException(e);

diff --git a/...ng/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2TreeIndex.java b/...ng/src/main/java/org/apache/ignite/internal/processors/query/h2/database/H2TreeIndex.java
@@ -50,12 +50,13 @@
 /**
  * H2 Index over {@link BPlusTree}.
  */
+@SuppressWarnings({"TypeMayBeWeakened", "unchecked"})
 public class H2TreeIndex extends GridH2IndexBase {
     /** Default value for {@code IGNITE_MAX_INDEX_PAYLOAD_SIZE} */
     public static final int IGNITE_MAX_INDEX_PAYLOAD_SIZE_DEFAULT = 10;
 
     /** */
-    private final H2Tree tree;
+    private final H2Tree[] segments;
 
     /** */
     private final List<InlineIndexHelper> inlineIdxs;
@@ -78,8 +79,11 @@ public H2TreeIndex(
         String name,
         boolean pk,
         List<IndexColumn> colsList,
-        int inlineSize
+        int inlineSize,
+        int segmentsCnt
     ) throws IgniteCheckedException {
+        assert segmentsCnt > 0 : segmentsCnt;
+
         this.cctx = cctx;
         IndexColumn[] cols = colsList.toArray(new IndexColumn[colsList.size()]);
 
@@ -95,21 +99,35 @@ public H2TreeIndex(
         if (cctx.affinityNode()) {
             IgniteCacheDatabaseSharedManager dbMgr = cctx.shared().database();
 
-            RootPage page = getMetaPage(name);
-
             inlineIdxs = getAvailableInlineColumns(cols);
 
-            tree = new H2Tree(name, cctx.offheap().reuseListForIndex(name), cctx.cacheId(),
+            segments = new H2Tree[segmentsCnt];
-                dbMgr.pageMemory(), cctx.shared().wal(), cctx.offheap().globalRemoveId(),
+
-                tbl.rowFactory(), page.pageId().pageId(), page.isAllocated(), cols, inlineIdxs, computeInlineSize(inlineIdxs, inlineSize)) {
+            for (int i = 0; i < segments.length; i++) {
-                @Override public int compareValues(Value v1, Value v2) {
+                RootPage page = getMetaPage(name, i);
-                    return v1 == v2 ? 0 : table.compareTypeSafe(v1, v2);
+
-                }
+                segments[i] = new H2Tree(
-            };
+                    name,
+                    cctx.offheap().reuseListForIndex(name),
+                    cctx.cacheId(),
+                    dbMgr.pageMemory(),
+                    cctx.shared().wal(),
+                    cctx.offheap().globalRemoveId(),
+                    tbl.rowFactory(),
+                    page.pageId().pageId(),
+                    page.isAllocated(),
+                    cols,
+                    inlineIdxs,
+                    computeInlineSize(inlineIdxs, inlineSize)) {
+                    @Override public int compareValues(Value v1, Value v2) {
+                        return v1 == v2 ? 0 : table.compareTypeSafe(v1, v2);
+                    }
+                };
+            }
         }
         else {
             // We need indexes on the client node, but index will not contain any data.
-            tree = null;
+            segments = null;
             inlineIdxs = null;
         }
 
@@ -123,9 +141,7 @@ public H2TreeIndex(
     private List<InlineIndexHelper> getAvailableInlineColumns(IndexColumn[] cols) {
         List<InlineIndexHelper> res = new ArrayList<>();
 
-        for (int i = 0; i < cols.length; i++) {
+        for (IndexColumn col : cols) {
-            IndexColumn col = cols[i];
-
             if (!InlineIndexHelper.AVAILABLE_TYPES.contains(col.column.getType()))
                 break;
 
@@ -154,6 +170,10 @@ private List<InlineIndexHelper> getAvailableInlineColumns(IndexColumn[] cols) {
                 p = f.forSpace(spaceName);
             }
 
+            int seg = threadLocalSegment();
+
+            H2Tree tree = treeForRead(seg);
+
             return new H2Cursor(tree.find(lower, upper), p);
         }
         catch (IgniteCheckedException e) {
@@ -164,6 +184,10 @@ private List<InlineIndexHelper> getAvailableInlineColumns(IndexColumn[] cols) {
     /** {@inheritDoc} */
     @Override public GridH2Row findOne(GridH2Row row) {
         try {
+            int seg = threadLocalSegment();
+
+            H2Tree tree = treeForRead(seg);
+
             return tree.findOne(row);
         }
         catch (IgniteCheckedException e) {
@@ -176,6 +200,10 @@ private List<InlineIndexHelper> getAvailableInlineColumns(IndexColumn[] cols) {
         try {
             InlineIndexHelper.setCurrentInlineIndexes(inlineIdxs);
 
+            int seg = threadLocalSegment();
+
+            H2Tree tree = treeForRead(seg);
+
             return tree.put(row);
         }
         catch (IgniteCheckedException e) {
@@ -191,6 +219,10 @@ private List<InlineIndexHelper> getAvailableInlineColumns(IndexColumn[] cols) {
         try {
             InlineIndexHelper.setCurrentInlineIndexes(inlineIdxs);
 
+            int seg = threadLocalSegment();
+
+            H2Tree tree = treeForRead(seg);
+
             return tree.putx(row);
         }
         catch (IgniteCheckedException e) {
@@ -205,6 +237,11 @@ private List<InlineIndexHelper> getAvailableInlineColumns(IndexColumn[] cols) {
     @Override public GridH2Row remove(SearchRow row) {
         try {
             InlineIndexHelper.setCurrentInlineIndexes(inlineIdxs);
+
+            int seg = threadLocalSegment();
+
+            H2Tree tree = treeForRead(seg);
+
             return tree.remove(row);
         }
         catch (IgniteCheckedException e) {
@@ -219,6 +256,11 @@ private List<InlineIndexHelper> getAvailableInlineColumns(IndexColumn[] cols) {
     @Override public void removex(SearchRow row) {
         try {
             InlineIndexHelper.setCurrentInlineIndexes(inlineIdxs);
+
+            int seg = threadLocalSegment();
+
+            H2Tree tree = treeForRead(seg);
+
             tree.removex(row);
         }
         catch (IgniteCheckedException e) {
@@ -271,9 +313,11 @@ private List<InlineIndexHelper> getAvailableInlineColumns(IndexColumn[] cols) {
     @Override public void destroy() {
         try {
             if (cctx.affinityNode()) {
-                tree.destroy();
+                for (H2Tree tree : segments) {
+                    tree.destroy();
 
-                cctx.offheap().dropRootPageForIndex(tree.getName());
+                    cctx.offheap().dropRootPageForIndex(tree.getName());
+                }
             }
         }
         catch (IgniteCheckedException e) {
@@ -286,17 +330,14 @@ private List<InlineIndexHelper> getAvailableInlineColumns(IndexColumn[] cols) {
 
     /** {@inheritDoc} */
     @Nullable @Override protected IgniteTree<SearchRow, GridH2Row> doTakeSnapshot() {
-        return tree;
+        int seg = threadLocalSegment();
-    }
 
-    /** {@inheritDoc} */
+        return treeForRead(seg);
-    protected IgniteTree<SearchRow, GridH2Row> treeForRead() {
-        return tree;
     }
 
     /** {@inheritDoc} */
-    @Override protected <K, V> IgniteTree<K, V> treeForRead(int segment) {
+    @Override protected H2Tree treeForRead(int segment) {
-        return (IgniteTree<K, V>)tree;
+        return segments[segment];
     }
 
     /** {@inheritDoc} */
@@ -306,9 +347,11 @@ protected IgniteTree<SearchRow, GridH2Row> treeForRead() {
         boolean includeFirst,
         @Nullable SearchRow last,
         IndexingQueryFilter filter) {
-        includeFirst &= first != null;
-
         try {
+            int seg = threadLocalSegment();
+
+            H2Tree tree = treeForRead(seg);
+
             GridCursor<GridH2Row> range = tree.find(first, last);
 
             if (range == null)
@@ -344,8 +387,7 @@ private int computeInlineSize(List<InlineIndexHelper> inlineIdxs, int cfgInlineS
 
             int size = 0;
 
-            for (int i = 0; i < inlineIdxs.size(); i++) {
+            for (InlineIndexHelper idxHelper : inlineIdxs) {
-                InlineIndexHelper idxHelper = inlineIdxs.get(i);
                 if (idxHelper.size() <= 0) {
                     size = propSize;
                     break;
@@ -363,9 +405,9 @@ private int computeInlineSize(List<InlineIndexHelper> inlineIdxs, int cfgInlineS
     /**
      * @param name Name.
      * @return RootPage for meta page.
-     * @throws IgniteCheckedException
+     * @throws IgniteCheckedException If failed.
      */
-    private RootPage getMetaPage(String name) throws IgniteCheckedException {
+    private RootPage getMetaPage(String name, int segIdx) throws IgniteCheckedException {
-        return cctx.offheap().rootPageForIndex(name);
+        return cctx.offheap().rootPageForIndex(name + "%" + segIdx);
     }
 }
diff --git a/...ing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2TreeIndex.java b/...ing/src/main/java/org/apache/ignite/internal/processors/query/h2/opt/GridH2TreeIndex.java
@@ -517,7 +517,7 @@ private ComparableRow(SearchRow row, int bias) {
                 while(cursor.next()) {
                     GridH2Row row = cursor.get();
 
-                    // Check for interruptions every 1000 iterations.
+                    // Check for interruptions every 1024 iterations.
                     if ((++j & 1024) == 0 && thread.isInterrupted())
                         throw new InterruptedException();