@Test public void testFindFragmentsToMerge() { List<DataSegmentFragment> allFragments = Lists.newArrayList(); KylinConfig config = getTestConfig(); config.setProperty("kylin.stream.segment-min-fragments", "5"); config.setProperty("kylin.stream.max-fragment-size-mb", "100"); segmentStore.latestCheckpointFragment = 1000; List<DataSegmentFragment> result = segmentStore.chooseFragmentsToMerge(config, allFragments); assertTrue(result.isEmpty()); for (int i = 0; i < 10; i++) { allFragments.add(new MockFragment(new FragmentId(i), 15)); } result = segmentStore.chooseFragmentsToMerge(config, allFragments); assertEquals(6, result.size()); allFragments.clear(); for (int i = 0; i < 50; i++) { allFragments.add(new MockFragment(new FragmentId(i), 1)); } result = segmentStore.chooseFragmentsToMerge(config, allFragments); assertEquals(46, result.size()); allFragments.clear(); allFragments.add(new MockFragment(new FragmentId(0), 100)); for (int i = 1; i < 10; i++) { allFragments.add(new MockFragment(new FragmentId(i), 15)); } result = segmentStore.chooseFragmentsToMerge(config, allFragments); assertEquals(6, result.size()); allFragments.clear(); allFragments.add(new MockFragment(new FragmentId(0, 5), 50)); for (int i = 6; i < 20; i++) { allFragments.add(new MockFragment(new FragmentId(i), 15)); } result = segmentStore.chooseFragmentsToMerge(config, allFragments); assertTrue(result.get(0).getFragmentId().equals(new FragmentId(6))); assertEquals(6, result.size()); }
protected List<DataSegmentFragment> chooseFragmentsToMerge(KylinConfig config, List<DataSegmentFragment> allFragments) { Collections.sort(allFragments); List<DataSegmentFragment> result = doChooseFragments(config, allFragments, true); return result; }
ColumnarSegmentStore implements IStreamingSegmentStore { protected List<DataSegmentFragment> chooseFragmentsToMerge(KylinConfig config, List<DataSegmentFragment> allFragments) { Collections.sort(allFragments); List<DataSegmentFragment> result = doChooseFragments(config, allFragments, true); return result; } }
ColumnarSegmentStore implements IStreamingSegmentStore { protected List<DataSegmentFragment> chooseFragmentsToMerge(KylinConfig config, List<DataSegmentFragment> allFragments) { Collections.sort(allFragments); List<DataSegmentFragment> result = doChooseFragments(config, allFragments, true); return result; } ColumnarSegmentStore(String baseStorePath, CubeInstance cubeInstance, String segmentName); }
ColumnarSegmentStore implements IStreamingSegmentStore { protected List<DataSegmentFragment> chooseFragmentsToMerge(KylinConfig config, List<DataSegmentFragment> allFragments) { Collections.sort(allFragments); List<DataSegmentFragment> result = doChooseFragments(config, allFragments, true); return result; } ColumnarSegmentStore(String baseStorePath, CubeInstance cubeInstance, String segmentName); @Override void init(); int addEvent(StreamingMessage event); @Override void addExternalDict(Map<TblColRef, Dictionary<String>> dictMap); @Override File getStorePath(); @Override Object checkpoint(); @Override void persist(); @Override void purge(); @Override void restoreFromCheckpoint(Object checkpoint); @Override String getSegmentName(); @Override StreamingCubeSegment.State getSegmentState(); @Override void setSegmentState(StreamingCubeSegment.State state); DataSegmentFragment createNewFragment(); List<DataSegmentFragment> getAllFragments(); @Override SegmentStoreStats getStoreStats(); SegmentMemoryStore getActiveMemoryStore(); @Override void search(final StreamingSearchContext searchContext, ResultCollector collector); void close(); }
ColumnarSegmentStore implements IStreamingSegmentStore { protected List<DataSegmentFragment> chooseFragmentsToMerge(KylinConfig config, List<DataSegmentFragment> allFragments) { Collections.sort(allFragments); List<DataSegmentFragment> result = doChooseFragments(config, allFragments, true); return result; } ColumnarSegmentStore(String baseStorePath, CubeInstance cubeInstance, String segmentName); @Override void init(); int addEvent(StreamingMessage event); @Override void addExternalDict(Map<TblColRef, Dictionary<String>> dictMap); @Override File getStorePath(); @Override Object checkpoint(); @Override void persist(); @Override void purge(); @Override void restoreFromCheckpoint(Object checkpoint); @Override String getSegmentName(); @Override StreamingCubeSegment.State getSegmentState(); @Override void setSegmentState(StreamingCubeSegment.State state); DataSegmentFragment createNewFragment(); List<DataSegmentFragment> getAllFragments(); @Override SegmentStoreStats getStoreStats(); SegmentMemoryStore getActiveMemoryStore(); @Override void search(final StreamingSearchContext searchContext, ResultCollector collector); void close();  }