kylin-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From shaofeng...@apache.org
Subject [2/3] kylin git commit: KYLIN-2059 Concurrent build issue in CubeManager.calculateToBeSegments()
Date Thu, 06 Oct 2016 06:46:10 GMT
KYLIN-2059 Concurrent build issue in CubeManager.calculateToBeSegments()

Project: http://git-wip-us.apache.org/repos/asf/kylin/repo
Commit: http://git-wip-us.apache.org/repos/asf/kylin/commit/334c2e09
Tree: http://git-wip-us.apache.org/repos/asf/kylin/tree/334c2e09
Diff: http://git-wip-us.apache.org/repos/asf/kylin/diff/334c2e09

Branch: refs/heads/master
Commit: 334c2e09ba443b8fdcb7d4bfe08ab8fbc0ac3fbe
Parents: aa51ce0
Author: shaofengshi <shaofengshi@apache.org>
Authored: Thu Sep 29 22:56:00 2016 +0800
Committer: shaofengshi <shaofengshi@apache.org>
Committed: Thu Oct 6 14:44:05 2016 +0800

----------------------------------------------------------------------
 .../apache/kylin/common/KylinConfigBase.java    |  4 +
 .../java/org/apache/kylin/cube/CubeManager.java | 77 ++++++++-------
 .../org/apache/kylin/cube/CubeManagerTest.java  | 99 ++++++++++++++++++++
 3 files changed, 145 insertions(+), 35 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/kylin/blob/334c2e09/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
----------------------------------------------------------------------
diff --git a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
index 838ef97..4d1639b 100644
--- a/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
+++ b/core-common/src/main/java/org/apache/kylin/common/KylinConfigBase.java
@@ -809,4 +809,8 @@ abstract public class KylinConfigBase implements Serializable {
     public int getMaxBuildingSegments() {
         return Integer.parseInt(getOptional("kylin.cube.building.segment.max", "2"));
     }
+
+    public void setMaxBuildingSegments(int maxBuildingSegments) {
+        setProperty("kylin.cube.building.segment.max", String.valueOf(maxBuildingSegments));
+    }
 }

http://git-wip-us.apache.org/repos/asf/kylin/blob/334c2e09/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java
----------------------------------------------------------------------
diff --git a/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java b/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java
index 962568c..d243f4d 100644
--- a/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java
+++ b/core-cube/src/main/java/org/apache/kylin/cube/CubeManager.java
@@ -135,7 +135,7 @@ public class CubeManager implements IRealizationProvider {
         logger.info("Initializing CubeManager with config " + config);
         this.config = config;
         this.cubeMap = new CaseInsensitiveStringCache<CubeInstance>(config, "cube");
-        
+
         // touch lower level metadata before registering my listener
         loadAllCubeInstance();
         Broadcaster.getInstance(config).registerListener(new CubeSyncListener(), "cube");
@@ -159,12 +159,12 @@ public class CubeManager implements IRealizationProvider {
         @Override
         public void onEntityChange(Broadcaster broadcaster, String entity, Event event, String
cacheKey) throws IOException {
             String cubeName = cacheKey;
-            
+
             if (event == Event.DROP)
                 removeCubeLocal(cubeName);
             else
                 reloadCubeLocal(cubeName);
-            
+
             for (ProjectInstance prj : ProjectManager.getInstance(config).findProjects(RealizationType.CUBE,
cubeName)) {
                 broadcaster.notifyProjectDataUpdate(prj.getName());
             }
@@ -615,7 +615,6 @@ public class CubeManager implements IRealizationProvider {
         return max;
     }
 
-
     private long calculateStartOffsetForAppendSegment(CubeInstance cube) {
         List<CubeSegment> existing = cube.getSegments();
         if (existing.isEmpty()) {
@@ -625,7 +624,6 @@ public class CubeManager implements IRealizationProvider {
         }
     }
 
-
     private long calculateStartDateForAppendSegment(CubeInstance cube) {
         List<CubeSegment> existing = cube.getSegments();
         if (existing.isEmpty()) {
@@ -728,7 +726,7 @@ public class CubeManager implements IRealizationProvider {
 
         List<CubeSegment> mergingSegs = Lists.newArrayList();
         if (buildingSegs.size() > 0) {
-            
+
             for (CubeSegment building : buildingSegs) {
                 // exclude those under-merging segs
                 for (CubeSegment ready : readySegs) {
@@ -760,27 +758,22 @@ public class CubeManager implements IRealizationProvider {
         return null;
     }
 
-    public void promoteNewlyBuiltSegments(CubeInstance cube, CubeSegment... newSegments)
throws IOException {
-        List<CubeSegment> tobe = calculateToBeSegments(cube);
+    public void promoteNewlyBuiltSegments(CubeInstance cube, CubeSegment newSegment) throws
IOException {
+        if (StringUtils.isBlank(newSegment.getStorageLocationIdentifier()))
+            throw new IllegalStateException("For cube " + cube + ", segment " + newSegment
+ " missing StorageLocationIdentifier");
 
-        for (CubeSegment seg : newSegments) {
-            if (tobe.contains(seg) == false)
-                throw new IllegalStateException("For cube " + cube + ", segment " + seg +
" is expected but not in the tobe " + tobe);
+        if (StringUtils.isBlank(newSegment.getLastBuildJobID()))
+            throw new IllegalStateException("For cube " + cube + ", segment " + newSegment
+ " missing LastBuildJobID");
 
-            if (StringUtils.isBlank(seg.getStorageLocationIdentifier()))
-                throw new IllegalStateException("For cube " + cube + ", segment " + seg +
" missing StorageLocationIdentifier");
+        if (isReady(newSegment) == true)
+            throw new IllegalStateException("For cube " + cube + ", segment " + newSegment
+ " state should be NEW but is READY");
 
-            if (StringUtils.isBlank(seg.getLastBuildJobID()))
-                throw new IllegalStateException("For cube " + cube + ", segment " + seg +
" missing LastBuildJobID");
+        List<CubeSegment> tobe = calculateToBeSegments(cube, newSegment);
 
-            seg.setStatus(SegmentStatusEnum.READY);
-        }
+        if (tobe.contains(newSegment) == false)
+            throw new IllegalStateException("For cube " + cube + ", segment " + newSegment
+ " is expected but not in the tobe " + tobe);
 
-        for (CubeSegment seg : tobe) {
-            if (isReady(seg) == false) {
-                logger.warn("For cube " + cube + ", segment " + seg + " isn't READY yet.");
-            }
-        }
+        newSegment.setStatus(SegmentStatusEnum.READY);
 
         List<CubeSegment> toRemoveSegs = Lists.newArrayList();
         for (CubeSegment segment : cube.getSegments()) {
@@ -788,14 +781,14 @@ public class CubeManager implements IRealizationProvider {
                 toRemoveSegs.add(segment);
         }
 
-        logger.info("Promoting cube " + cube + ", new segments " + Arrays.toString(newSegments)
+ ", to remove segments " + toRemoveSegs);
+        logger.info("Promoting cube " + cube + ", new segment " + newSegment + ", to remove
segments " + toRemoveSegs);
 
         CubeUpdate cubeBuilder = new CubeUpdate(cube);
-        cubeBuilder.setToRemoveSegs(toRemoveSegs.toArray(new CubeSegment[toRemoveSegs.size()])).setToUpdateSegs(newSegments).setStatus(RealizationStatusEnum.READY);
+        cubeBuilder.setToRemoveSegs(toRemoveSegs.toArray(new CubeSegment[toRemoveSegs.size()])).setToUpdateSegs(newSegment).setStatus(RealizationStatusEnum.READY);
         updateCube(cubeBuilder);
     }
 
-    public void validateNewSegments(CubeInstance cube, CubeSegment... newSegments) {
+    public void validateNewSegments(CubeInstance cube, CubeSegment newSegments) {
         List<CubeSegment> tobe = calculateToBeSegments(cube, newSegments);
         List<CubeSegment> newList = Arrays.asList(newSegments);
         if (tobe.containsAll(newList) == false) {
@@ -809,11 +802,12 @@ public class CubeManager implements IRealizationProvider {
      * - Favors new segments over the old
      * - Favors big segments over the small
      */
-    private List<CubeSegment> calculateToBeSegments(CubeInstance cube, CubeSegment...
newSegments) {
+    private List<CubeSegment> calculateToBeSegments(CubeInstance cube, CubeSegment
newSegments) {
 
         List<CubeSegment> tobe = Lists.newArrayList(cube.getSegments());
-        if (newSegments != null)
-            tobe.addAll(Arrays.asList(newSegments));
+        if (newSegments != null && !tobe.contains(newSegments)) {
+            tobe.add(newSegments);
+        }
         if (tobe.size() == 0)
             return tobe;
 
@@ -849,13 +843,17 @@ public class CubeManager implements IRealizationProvider {
                     } else {
                         tobe.remove(j);
                     }
-                } else if (isNew(is)) {
-                    // otherwise, favor the new segment
-                    tobe.remove(j);
+                    continue;
                 } else {
-                    tobe.remove(i);
+                    // otherwise, favor the new segment
+                    if (isNew(is) && is.equals(newSegments)) {
+                        tobe.remove(j);
+                        continue;
+                    } else if (js.equals(newSegments)) {
+                        tobe.remove(i);
+                        continue;
+                    }
                 }
-                continue;
             }
 
             // if i, j in sequence
@@ -865,8 +863,17 @@ public class CubeManager implements IRealizationProvider {
                 continue;
             }
 
-            // seems j not fitting
-            tobe.remove(j);
+            // js can be covered by is
+            if (is.equals(newSegments)) {
+                // seems j not fitting
+                tobe.remove(j);
+                continue;
+            } else {
+                i++;
+                j++;
+                continue;
+            }
+
         }
 
         return tobe;

http://git-wip-us.apache.org/repos/asf/kylin/blob/334c2e09/core-cube/src/test/java/org/apache/kylin/cube/CubeManagerTest.java
----------------------------------------------------------------------
diff --git a/core-cube/src/test/java/org/apache/kylin/cube/CubeManagerTest.java b/core-cube/src/test/java/org/apache/kylin/cube/CubeManagerTest.java
index 49bb128..e63fe99 100644
--- a/core-cube/src/test/java/org/apache/kylin/cube/CubeManagerTest.java
+++ b/core-cube/src/test/java/org/apache/kylin/cube/CubeManagerTest.java
@@ -19,6 +19,7 @@
 package org.apache.kylin.cube;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
@@ -123,6 +124,104 @@ public class CubeManagerTest extends LocalFileMetadataTestCase {
 
     }
 
+
+    @Test
+    public void testConcurrentBuildAndMerge() throws Exception {
+        CubeManager mgr = CubeManager.getInstance(getTestConfig());
+        CubeInstance cube = mgr.getCube("test_kylin_cube_with_slr_empty");
+        getTestConfig().setMaxBuildingSegments(10);
+        // no segment at first
+        assertEquals(0, cube.getSegments().size());
+
+        // append first
+        CubeSegment seg1 = mgr.appendSegment(cube, 0, 0, 0, 1000);
+        seg1.setStatus(SegmentStatusEnum.READY);
+
+        CubeSegment seg2 = mgr.appendSegment(cube, 0, 0, 1000, 2000);
+        seg2.setStatus(SegmentStatusEnum.READY);
+
+
+        CubeSegment seg3 = mgr.mergeSegments(cube, 0, 0, 0000, 2000, true);
+        seg3.setStatus(SegmentStatusEnum.NEW);
+
+
+        CubeSegment seg4 = mgr.appendSegment(cube, 0, 0, 2000, 3000);
+        seg4.setStatus(SegmentStatusEnum.NEW);
+        seg4.setLastBuildJobID("test");
+        seg4.setStorageLocationIdentifier("test");
+
+        CubeSegment seg5 = mgr.appendSegment(cube, 0, 0, 3000, 4000);
+        seg5.setStatus(SegmentStatusEnum.READY);
+
+        CubeUpdate cubeBuilder = new CubeUpdate(cube);
+
+        mgr.updateCube(cubeBuilder);
+
+
+        mgr.promoteNewlyBuiltSegments(cube, seg4);
+
+        assertTrue(cube.getSegments().size() == 5);
+
+        assertTrue(cube.getSegmentById(seg1.getUuid()) != null && cube.getSegmentById(seg1.getUuid()).getStatus()
== SegmentStatusEnum.READY);
+        assertTrue(cube.getSegmentById(seg2.getUuid()) != null && cube.getSegmentById(seg2.getUuid()).getStatus()
== SegmentStatusEnum.READY);
+        assertTrue(cube.getSegmentById(seg3.getUuid()) != null && cube.getSegmentById(seg3.getUuid()).getStatus()
== SegmentStatusEnum.NEW);
+        assertTrue(cube.getSegmentById(seg4.getUuid()) != null && cube.getSegmentById(seg4.getUuid()).getStatus()
== SegmentStatusEnum.READY);
+        assertTrue(cube.getSegmentById(seg5.getUuid()) != null && cube.getSegmentById(seg5.getUuid()).getStatus()
== SegmentStatusEnum.READY);
+
+    }
+
+
+    @Test
+    public void testConcurrentMergeAndMerge() throws Exception {
+        CubeManager mgr = CubeManager.getInstance(getTestConfig());
+        getTestConfig().setMaxBuildingSegments(10);
+        CubeInstance cube = mgr.getCube("test_kylin_cube_with_slr_empty");
+
+        // no segment at first
+        assertEquals(0, cube.getSegments().size());
+
+        // append first
+        CubeSegment seg1 = mgr.appendSegment(cube, 0, 0, 0, 1000);
+        seg1.setStatus(SegmentStatusEnum.READY);
+
+        CubeSegment seg2 = mgr.appendSegment(cube, 0, 0, 1000, 2000);
+        seg2.setStatus(SegmentStatusEnum.READY);
+
+        CubeSegment seg3 = mgr.appendSegment(cube, 0, 0, 2000, 3000);
+        seg3.setStatus(SegmentStatusEnum.READY);
+
+        CubeSegment seg4 = mgr.appendSegment(cube, 0, 0, 3000, 4000);
+        seg4.setStatus(SegmentStatusEnum.READY);
+
+
+
+        CubeSegment merge1 = mgr.mergeSegments(cube, 0, 0, 0, 2000, true);
+        merge1.setStatus(SegmentStatusEnum.NEW);
+        merge1.setLastBuildJobID("test");
+        merge1.setStorageLocationIdentifier("test");
+
+        CubeSegment merge2 = mgr.mergeSegments(cube, 0, 0, 2000, 4000, true);
+        merge2.setStatus(SegmentStatusEnum.NEW);
+        merge2.setLastBuildJobID("test");
+        merge2.setStorageLocationIdentifier("test");
+
+        CubeUpdate cubeBuilder = new CubeUpdate(cube);
+        mgr.updateCube(cubeBuilder);
+
+
+        mgr.promoteNewlyBuiltSegments(cube, merge1);
+
+        assertTrue(cube.getSegments().size() == 4);
+
+        assertTrue(cube.getSegmentById(seg1.getUuid()) == null);
+        assertTrue(cube.getSegmentById(seg2.getUuid()) == null);
+        assertTrue(cube.getSegmentById(merge1.getUuid()) != null && cube.getSegmentById(merge1.getUuid()).getStatus()
== SegmentStatusEnum.READY);
+        assertTrue(cube.getSegmentById(seg3.getUuid()) != null && cube.getSegmentById(seg3.getUuid()).getStatus()
== SegmentStatusEnum.READY);
+        assertTrue(cube.getSegmentById(seg4.getUuid()) != null && cube.getSegmentById(seg4.getUuid()).getStatus()
== SegmentStatusEnum.READY);
+        assertTrue(cube.getSegmentById(merge2.getUuid()) != null && cube.getSegmentById(merge2.getUuid()).getStatus()
== SegmentStatusEnum.NEW);
+
+    }
+
     @Test
     public void testGetAllCubes() throws Exception {
         final ResourceStore store = ResourceStore.getStore(getTestConfig());


Mime
View raw message