@Test public void testMerge2() throws Exception { int fragmentNum = 4; int eventCntPerMin = 100000; StreamingDataSimulator simulator = new StreamingDataSimulator(getTestCardinalityMap(), eventCntPerMin); List<DataSegmentFragment> fragments = createFragmentFiles(5, fragmentNum, simulator); int originRowCnt = fragmentNum * 50000; FragmentsMergeResult mergeResult = fragmentFilesMerger.merge(fragments); File mergedFragmentMetaFile = mergeResult.getMergedFragmentMetaFile(); File mergedFragmentDataFile = mergeResult.getMergedFragmentDataFile(); FragmentMetaInfo fragmentMetaInfo = JsonUtil.readValue(mergedFragmentMetaFile, FragmentMetaInfo.class); assertEquals(160000, fragmentMetaInfo.getNumberOfRows()); assertEquals(originRowCnt, fragmentMetaInfo.getOriginNumOfRows()); FragmentData fragmentData = new FragmentData(fragmentMetaInfo, mergedFragmentDataFile); Set<TblColRef> dims = testHelper.simulateDimensions("STREAMING_V2_TABLE.MINUTE_START"); TupleFilter siteFilter = null; Set<TblColRef> groups = testHelper.simulateDimensions("STREAMING_V2_TABLE.MINUTE_START"); Set<FunctionDesc> metrics = Sets.newHashSet(testHelper.simulateCountMetric()); StreamingQueryProfile.set(new StreamingQueryProfile("test-query-id", System.currentTimeMillis())); DataSegmentFragment fragment = new DataSegmentFragment(baseStorePath, cubeName, segmentName, new FragmentId(0)); FragmentFileSearcher fragmentFileSearcher = new FragmentFileSearcher(fragment, fragmentData); StreamingSearchContext searchRequest = new StreamingSearchContext(parsedStreamingCubeInfo.cubeDesc, dims, groups, metrics, siteFilter, null); ResultCollector resultCollector = new SingleThreadResultCollector(); fragmentFileSearcher.search(searchRequest, resultCollector); int rowCnt = 0; int totalOriginCnt = 0; for (Record record : resultCollector) { rowCnt++; long everyMinuteCnt = (Long) record.getMetrics()[0]; assertEquals(eventCntPerMin, (int) everyMinuteCnt); System.out.println(record); totalOriginCnt += everyMinuteCnt; } assertEquals(2, rowCnt); assertEquals(originRowCnt, totalOriginCnt); }
public FragmentsMergeResult merge(List<DataSegmentFragment> fragmentList) throws IOException { if (!mergeWorkingDirectory.exists()) { mergeWorkingDirectory.mkdirs(); } else { logger.info("clean the merge working dir:{}", mergeWorkingDirectory.getAbsolutePath()); FileUtils.cleanDirectory(mergeWorkingDirectory); } Collections.sort(fragmentList); FragmentId mergedFragmentId = new FragmentId(fragmentList.get(0).getFragmentId().getStartId(), fragmentList.get(fragmentList.size() - 1).getFragmentId().getEndId()); List<FragmentData> fragmentDataList = Lists.newArrayList(); Map<TblColRef, List<Dictionary<String>>> dimDictListMap = Maps.newHashMap(); Map<FragmentId, Map<TblColRef, Dictionary<String>>> fragmentDictionaryMaps = Maps.newHashMap(); List<Long> additionalCuboidsToMerge = null; long minMergedFragmentEventTime = Long.MAX_VALUE; long maxMergedFragmentEventTime = 0; long originNumOfRows = 0; for (DataSegmentFragment fragment : fragmentList) { FragmentData fragmentData = ColumnarStoreCache.getInstance().startReadFragmentData(fragment); FragmentMetaInfo fragmentMetaInfo = fragmentData.getFragmentMetaInfo(); long fragmentMinTime = fragmentMetaInfo.getMinEventTime(); long fragmentMaxTime = fragmentMetaInfo.getMaxEventTime(); originNumOfRows += fragmentMetaInfo.getOriginNumOfRows(); if (fragmentMinTime < minMergedFragmentEventTime) { minMergedFragmentEventTime = fragmentMinTime; } if (fragmentMaxTime > maxMergedFragmentEventTime) { maxMergedFragmentEventTime = fragmentMaxTime; } if (additionalCuboidsToMerge == null) { Map<String, CuboidMetaInfo> cuboidMetaInfoMap = fragmentMetaInfo.getCuboidMetaInfoMap(); if (cuboidMetaInfoMap != null) { additionalCuboidsToMerge = Lists.transform(Lists.newArrayList(cuboidMetaInfoMap.keySet()), new Function<String, Long>() { @Nullable @Override public Long apply(@Nullable String input) { return Long.valueOf(input); } }); } else { additionalCuboidsToMerge = Lists.newArrayList(); } } fragmentDataList.add(fragmentData); Map<TblColRef, Dictionary<String>> dictionaryMap = fragmentData .getDimensionDictionaries(parsedCubeInfo.dimensionsUseDictEncoding); fragmentDictionaryMaps.put(fragment.getFragmentId(), dictionaryMap); for (Entry<TblColRef, Dictionary<String>> dimDictEntry : dictionaryMap.entrySet()) { List<Dictionary<String>> dictionaryList = dimDictListMap.get(dimDictEntry.getKey()); if (dictionaryList == null) { dictionaryList = Lists.newArrayList(); dimDictListMap.put(dimDictEntry.getKey(), dictionaryList); } dictionaryList.add(dimDictEntry.getValue()); } } File mergedFragmentDataFile = new File(mergeWorkingDirectory, mergedFragmentId + Constants.DATA_FILE_SUFFIX); File mergedFragmentMetaFile = new File(mergeWorkingDirectory, mergedFragmentId + Constants.META_FILE_SUFFIX); try { FragmentMetaInfo mergedFragmentMeta = new FragmentMetaInfo(); CountingOutputStream fragmentDataOutput = new CountingOutputStream( new BufferedOutputStream(FileUtils.openOutputStream(mergedFragmentDataFile))); Map<TblColRef, Dictionary<String>> mergedDictMap = mergeAndPersistDictionaries(mergedFragmentMeta, dimDictListMap, fragmentDataOutput); logger.info("merge basic cuboid"); CuboidMetaInfo basicCuboidMeta = mergeAndPersistCuboidData(fragmentDataList, fragmentDictionaryMaps, mergedDictMap, fragmentDataOutput, parsedCubeInfo.basicCuboid.getId()); mergedFragmentMeta.setBasicCuboidMetaInfo(basicCuboidMeta); long totalRowCnt = basicCuboidMeta.getNumberOfRows(); Map<String, CuboidMetaInfo> cuboidMetaInfoMap = Maps.newHashMap(); for (Long cuboidId : additionalCuboidsToMerge) { logger.info("merge cuboid:{}", cuboidId); CuboidMetaInfo cuboidMeta = mergeAndPersistCuboidData(fragmentDataList, fragmentDictionaryMaps, mergedDictMap, fragmentDataOutput, cuboidId); cuboidMetaInfoMap.put(String.valueOf(cuboidId), cuboidMeta); totalRowCnt += cuboidMeta.getNumberOfRows(); } mergedFragmentMeta.setMaxEventTime(maxMergedFragmentEventTime); mergedFragmentMeta.setMinEventTime(minMergedFragmentEventTime); mergedFragmentMeta.setCuboidMetaInfoMap(cuboidMetaInfoMap); mergedFragmentMeta.setFragmentId(mergedFragmentId.toString()); mergedFragmentMeta.setNumberOfRows(totalRowCnt); mergedFragmentMeta.setOriginNumOfRows(originNumOfRows); fragmentDataOutput.flush(); fragmentDataOutput.close(); FileOutputStream metaOutputStream = FileUtils.openOutputStream(mergedFragmentMetaFile); JsonUtil.writeValueIndent(metaOutputStream, mergedFragmentMeta); metaOutputStream.flush(); metaOutputStream.close(); } finally { for (DataSegmentFragment fragment : fragmentList) { ColumnarStoreCache.getInstance().finishReadFragmentData(fragment); } } FragmentsMergeResult result = new FragmentsMergeResult(fragmentList, mergedFragmentId, mergedFragmentMetaFile, mergedFragmentDataFile); return result; }
FragmentFilesMerger { public FragmentsMergeResult merge(List<DataSegmentFragment> fragmentList) throws IOException { if (!mergeWorkingDirectory.exists()) { mergeWorkingDirectory.mkdirs(); } else { logger.info("clean the merge working dir:{}", mergeWorkingDirectory.getAbsolutePath()); FileUtils.cleanDirectory(mergeWorkingDirectory); } Collections.sort(fragmentList); FragmentId mergedFragmentId = new FragmentId(fragmentList.get(0).getFragmentId().getStartId(), fragmentList.get(fragmentList.size() - 1).getFragmentId().getEndId()); List<FragmentData> fragmentDataList = Lists.newArrayList(); Map<TblColRef, List<Dictionary<String>>> dimDictListMap = Maps.newHashMap(); Map<FragmentId, Map<TblColRef, Dictionary<String>>> fragmentDictionaryMaps = Maps.newHashMap(); List<Long> additionalCuboidsToMerge = null; long minMergedFragmentEventTime = Long.MAX_VALUE; long maxMergedFragmentEventTime = 0; long originNumOfRows = 0; for (DataSegmentFragment fragment : fragmentList) { FragmentData fragmentData = ColumnarStoreCache.getInstance().startReadFragmentData(fragment); FragmentMetaInfo fragmentMetaInfo = fragmentData.getFragmentMetaInfo(); long fragmentMinTime = fragmentMetaInfo.getMinEventTime(); long fragmentMaxTime = fragmentMetaInfo.getMaxEventTime(); originNumOfRows += fragmentMetaInfo.getOriginNumOfRows(); if (fragmentMinTime < minMergedFragmentEventTime) { minMergedFragmentEventTime = fragmentMinTime; } if (fragmentMaxTime > maxMergedFragmentEventTime) { maxMergedFragmentEventTime = fragmentMaxTime; } if (additionalCuboidsToMerge == null) { Map<String, CuboidMetaInfo> cuboidMetaInfoMap = fragmentMetaInfo.getCuboidMetaInfoMap(); if (cuboidMetaInfoMap != null) { additionalCuboidsToMerge = Lists.transform(Lists.newArrayList(cuboidMetaInfoMap.keySet()), new Function<String, Long>() { @Nullable @Override public Long apply(@Nullable String input) { return Long.valueOf(input); } }); } else { additionalCuboidsToMerge = Lists.newArrayList(); } } fragmentDataList.add(fragmentData); Map<TblColRef, Dictionary<String>> dictionaryMap = fragmentData .getDimensionDictionaries(parsedCubeInfo.dimensionsUseDictEncoding); fragmentDictionaryMaps.put(fragment.getFragmentId(), dictionaryMap); for (Entry<TblColRef, Dictionary<String>> dimDictEntry : dictionaryMap.entrySet()) { List<Dictionary<String>> dictionaryList = dimDictListMap.get(dimDictEntry.getKey()); if (dictionaryList == null) { dictionaryList = Lists.newArrayList(); dimDictListMap.put(dimDictEntry.getKey(), dictionaryList); } dictionaryList.add(dimDictEntry.getValue()); } } File mergedFragmentDataFile = new File(mergeWorkingDirectory, mergedFragmentId + Constants.DATA_FILE_SUFFIX); File mergedFragmentMetaFile = new File(mergeWorkingDirectory, mergedFragmentId + Constants.META_FILE_SUFFIX); try { FragmentMetaInfo mergedFragmentMeta = new FragmentMetaInfo(); CountingOutputStream fragmentDataOutput = new CountingOutputStream( new BufferedOutputStream(FileUtils.openOutputStream(mergedFragmentDataFile))); Map<TblColRef, Dictionary<String>> mergedDictMap = mergeAndPersistDictionaries(mergedFragmentMeta, dimDictListMap, fragmentDataOutput); logger.info("merge basic cuboid"); CuboidMetaInfo basicCuboidMeta = mergeAndPersistCuboidData(fragmentDataList, fragmentDictionaryMaps, mergedDictMap, fragmentDataOutput, parsedCubeInfo.basicCuboid.getId()); mergedFragmentMeta.setBasicCuboidMetaInfo(basicCuboidMeta); long totalRowCnt = basicCuboidMeta.getNumberOfRows(); Map<String, CuboidMetaInfo> cuboidMetaInfoMap = Maps.newHashMap(); for (Long cuboidId : additionalCuboidsToMerge) { logger.info("merge cuboid:{}", cuboidId); CuboidMetaInfo cuboidMeta = mergeAndPersistCuboidData(fragmentDataList, fragmentDictionaryMaps, mergedDictMap, fragmentDataOutput, cuboidId); cuboidMetaInfoMap.put(String.valueOf(cuboidId), cuboidMeta); totalRowCnt += cuboidMeta.getNumberOfRows(); } mergedFragmentMeta.setMaxEventTime(maxMergedFragmentEventTime); mergedFragmentMeta.setMinEventTime(minMergedFragmentEventTime); mergedFragmentMeta.setCuboidMetaInfoMap(cuboidMetaInfoMap); mergedFragmentMeta.setFragmentId(mergedFragmentId.toString()); mergedFragmentMeta.setNumberOfRows(totalRowCnt); mergedFragmentMeta.setOriginNumOfRows(originNumOfRows); fragmentDataOutput.flush(); fragmentDataOutput.close(); FileOutputStream metaOutputStream = FileUtils.openOutputStream(mergedFragmentMetaFile); JsonUtil.writeValueIndent(metaOutputStream, mergedFragmentMeta); metaOutputStream.flush(); metaOutputStream.close(); } finally { for (DataSegmentFragment fragment : fragmentList) { ColumnarStoreCache.getInstance().finishReadFragmentData(fragment); } } FragmentsMergeResult result = new FragmentsMergeResult(fragmentList, mergedFragmentId, mergedFragmentMetaFile, mergedFragmentDataFile); return result; } }
FragmentFilesMerger { public FragmentsMergeResult merge(List<DataSegmentFragment> fragmentList) throws IOException { if (!mergeWorkingDirectory.exists()) { mergeWorkingDirectory.mkdirs(); } else { logger.info("clean the merge working dir:{}", mergeWorkingDirectory.getAbsolutePath()); FileUtils.cleanDirectory(mergeWorkingDirectory); } Collections.sort(fragmentList); FragmentId mergedFragmentId = new FragmentId(fragmentList.get(0).getFragmentId().getStartId(), fragmentList.get(fragmentList.size() - 1).getFragmentId().getEndId()); List<FragmentData> fragmentDataList = Lists.newArrayList(); Map<TblColRef, List<Dictionary<String>>> dimDictListMap = Maps.newHashMap(); Map<FragmentId, Map<TblColRef, Dictionary<String>>> fragmentDictionaryMaps = Maps.newHashMap(); List<Long> additionalCuboidsToMerge = null; long minMergedFragmentEventTime = Long.MAX_VALUE; long maxMergedFragmentEventTime = 0; long originNumOfRows = 0; for (DataSegmentFragment fragment : fragmentList) { FragmentData fragmentData = ColumnarStoreCache.getInstance().startReadFragmentData(fragment); FragmentMetaInfo fragmentMetaInfo = fragmentData.getFragmentMetaInfo(); long fragmentMinTime = fragmentMetaInfo.getMinEventTime(); long fragmentMaxTime = fragmentMetaInfo.getMaxEventTime(); originNumOfRows += fragmentMetaInfo.getOriginNumOfRows(); if (fragmentMinTime < minMergedFragmentEventTime) { minMergedFragmentEventTime = fragmentMinTime; } if (fragmentMaxTime > maxMergedFragmentEventTime) { maxMergedFragmentEventTime = fragmentMaxTime; } if (additionalCuboidsToMerge == null) { Map<String, CuboidMetaInfo> cuboidMetaInfoMap = fragmentMetaInfo.getCuboidMetaInfoMap(); if (cuboidMetaInfoMap != null) { additionalCuboidsToMerge = Lists.transform(Lists.newArrayList(cuboidMetaInfoMap.keySet()), new Function<String, Long>() { @Nullable @Override public Long apply(@Nullable String input) { return Long.valueOf(input); } }); } else { additionalCuboidsToMerge = Lists.newArrayList(); } } fragmentDataList.add(fragmentData); Map<TblColRef, Dictionary<String>> dictionaryMap = fragmentData .getDimensionDictionaries(parsedCubeInfo.dimensionsUseDictEncoding); fragmentDictionaryMaps.put(fragment.getFragmentId(), dictionaryMap); for (Entry<TblColRef, Dictionary<String>> dimDictEntry : dictionaryMap.entrySet()) { List<Dictionary<String>> dictionaryList = dimDictListMap.get(dimDictEntry.getKey()); if (dictionaryList == null) { dictionaryList = Lists.newArrayList(); dimDictListMap.put(dimDictEntry.getKey(), dictionaryList); } dictionaryList.add(dimDictEntry.getValue()); } } File mergedFragmentDataFile = new File(mergeWorkingDirectory, mergedFragmentId + Constants.DATA_FILE_SUFFIX); File mergedFragmentMetaFile = new File(mergeWorkingDirectory, mergedFragmentId + Constants.META_FILE_SUFFIX); try { FragmentMetaInfo mergedFragmentMeta = new FragmentMetaInfo(); CountingOutputStream fragmentDataOutput = new CountingOutputStream( new BufferedOutputStream(FileUtils.openOutputStream(mergedFragmentDataFile))); Map<TblColRef, Dictionary<String>> mergedDictMap = mergeAndPersistDictionaries(mergedFragmentMeta, dimDictListMap, fragmentDataOutput); logger.info("merge basic cuboid"); CuboidMetaInfo basicCuboidMeta = mergeAndPersistCuboidData(fragmentDataList, fragmentDictionaryMaps, mergedDictMap, fragmentDataOutput, parsedCubeInfo.basicCuboid.getId()); mergedFragmentMeta.setBasicCuboidMetaInfo(basicCuboidMeta); long totalRowCnt = basicCuboidMeta.getNumberOfRows(); Map<String, CuboidMetaInfo> cuboidMetaInfoMap = Maps.newHashMap(); for (Long cuboidId : additionalCuboidsToMerge) { logger.info("merge cuboid:{}", cuboidId); CuboidMetaInfo cuboidMeta = mergeAndPersistCuboidData(fragmentDataList, fragmentDictionaryMaps, mergedDictMap, fragmentDataOutput, cuboidId); cuboidMetaInfoMap.put(String.valueOf(cuboidId), cuboidMeta); totalRowCnt += cuboidMeta.getNumberOfRows(); } mergedFragmentMeta.setMaxEventTime(maxMergedFragmentEventTime); mergedFragmentMeta.setMinEventTime(minMergedFragmentEventTime); mergedFragmentMeta.setCuboidMetaInfoMap(cuboidMetaInfoMap); mergedFragmentMeta.setFragmentId(mergedFragmentId.toString()); mergedFragmentMeta.setNumberOfRows(totalRowCnt); mergedFragmentMeta.setOriginNumOfRows(originNumOfRows); fragmentDataOutput.flush(); fragmentDataOutput.close(); FileOutputStream metaOutputStream = FileUtils.openOutputStream(mergedFragmentMetaFile); JsonUtil.writeValueIndent(metaOutputStream, mergedFragmentMeta); metaOutputStream.flush(); metaOutputStream.close(); } finally { for (DataSegmentFragment fragment : fragmentList) { ColumnarStoreCache.getInstance().finishReadFragmentData(fragment); } } FragmentsMergeResult result = new FragmentsMergeResult(fragmentList, mergedFragmentId, mergedFragmentMetaFile, mergedFragmentDataFile); return result; } FragmentFilesMerger(ParsedStreamingCubeInfo parsedCubeInfo, File segmentFolder); }
FragmentFilesMerger { public FragmentsMergeResult merge(List<DataSegmentFragment> fragmentList) throws IOException { if (!mergeWorkingDirectory.exists()) { mergeWorkingDirectory.mkdirs(); } else { logger.info("clean the merge working dir:{}", mergeWorkingDirectory.getAbsolutePath()); FileUtils.cleanDirectory(mergeWorkingDirectory); } Collections.sort(fragmentList); FragmentId mergedFragmentId = new FragmentId(fragmentList.get(0).getFragmentId().getStartId(), fragmentList.get(fragmentList.size() - 1).getFragmentId().getEndId()); List<FragmentData> fragmentDataList = Lists.newArrayList(); Map<TblColRef, List<Dictionary<String>>> dimDictListMap = Maps.newHashMap(); Map<FragmentId, Map<TblColRef, Dictionary<String>>> fragmentDictionaryMaps = Maps.newHashMap(); List<Long> additionalCuboidsToMerge = null; long minMergedFragmentEventTime = Long.MAX_VALUE; long maxMergedFragmentEventTime = 0; long originNumOfRows = 0; for (DataSegmentFragment fragment : fragmentList) { FragmentData fragmentData = ColumnarStoreCache.getInstance().startReadFragmentData(fragment); FragmentMetaInfo fragmentMetaInfo = fragmentData.getFragmentMetaInfo(); long fragmentMinTime = fragmentMetaInfo.getMinEventTime(); long fragmentMaxTime = fragmentMetaInfo.getMaxEventTime(); originNumOfRows += fragmentMetaInfo.getOriginNumOfRows(); if (fragmentMinTime < minMergedFragmentEventTime) { minMergedFragmentEventTime = fragmentMinTime; } if (fragmentMaxTime > maxMergedFragmentEventTime) { maxMergedFragmentEventTime = fragmentMaxTime; } if (additionalCuboidsToMerge == null) { Map<String, CuboidMetaInfo> cuboidMetaInfoMap = fragmentMetaInfo.getCuboidMetaInfoMap(); if (cuboidMetaInfoMap != null) { additionalCuboidsToMerge = Lists.transform(Lists.newArrayList(cuboidMetaInfoMap.keySet()), new Function<String, Long>() { @Nullable @Override public Long apply(@Nullable String input) { return Long.valueOf(input); } }); } else { additionalCuboidsToMerge = Lists.newArrayList(); } } fragmentDataList.add(fragmentData); Map<TblColRef, Dictionary<String>> dictionaryMap = fragmentData .getDimensionDictionaries(parsedCubeInfo.dimensionsUseDictEncoding); fragmentDictionaryMaps.put(fragment.getFragmentId(), dictionaryMap); for (Entry<TblColRef, Dictionary<String>> dimDictEntry : dictionaryMap.entrySet()) { List<Dictionary<String>> dictionaryList = dimDictListMap.get(dimDictEntry.getKey()); if (dictionaryList == null) { dictionaryList = Lists.newArrayList(); dimDictListMap.put(dimDictEntry.getKey(), dictionaryList); } dictionaryList.add(dimDictEntry.getValue()); } } File mergedFragmentDataFile = new File(mergeWorkingDirectory, mergedFragmentId + Constants.DATA_FILE_SUFFIX); File mergedFragmentMetaFile = new File(mergeWorkingDirectory, mergedFragmentId + Constants.META_FILE_SUFFIX); try { FragmentMetaInfo mergedFragmentMeta = new FragmentMetaInfo(); CountingOutputStream fragmentDataOutput = new CountingOutputStream( new BufferedOutputStream(FileUtils.openOutputStream(mergedFragmentDataFile))); Map<TblColRef, Dictionary<String>> mergedDictMap = mergeAndPersistDictionaries(mergedFragmentMeta, dimDictListMap, fragmentDataOutput); logger.info("merge basic cuboid"); CuboidMetaInfo basicCuboidMeta = mergeAndPersistCuboidData(fragmentDataList, fragmentDictionaryMaps, mergedDictMap, fragmentDataOutput, parsedCubeInfo.basicCuboid.getId()); mergedFragmentMeta.setBasicCuboidMetaInfo(basicCuboidMeta); long totalRowCnt = basicCuboidMeta.getNumberOfRows(); Map<String, CuboidMetaInfo> cuboidMetaInfoMap = Maps.newHashMap(); for (Long cuboidId : additionalCuboidsToMerge) { logger.info("merge cuboid:{}", cuboidId); CuboidMetaInfo cuboidMeta = mergeAndPersistCuboidData(fragmentDataList, fragmentDictionaryMaps, mergedDictMap, fragmentDataOutput, cuboidId); cuboidMetaInfoMap.put(String.valueOf(cuboidId), cuboidMeta); totalRowCnt += cuboidMeta.getNumberOfRows(); } mergedFragmentMeta.setMaxEventTime(maxMergedFragmentEventTime); mergedFragmentMeta.setMinEventTime(minMergedFragmentEventTime); mergedFragmentMeta.setCuboidMetaInfoMap(cuboidMetaInfoMap); mergedFragmentMeta.setFragmentId(mergedFragmentId.toString()); mergedFragmentMeta.setNumberOfRows(totalRowCnt); mergedFragmentMeta.setOriginNumOfRows(originNumOfRows); fragmentDataOutput.flush(); fragmentDataOutput.close(); FileOutputStream metaOutputStream = FileUtils.openOutputStream(mergedFragmentMetaFile); JsonUtil.writeValueIndent(metaOutputStream, mergedFragmentMeta); metaOutputStream.flush(); metaOutputStream.close(); } finally { for (DataSegmentFragment fragment : fragmentList) { ColumnarStoreCache.getInstance().finishReadFragmentData(fragment); } } FragmentsMergeResult result = new FragmentsMergeResult(fragmentList, mergedFragmentId, mergedFragmentMetaFile, mergedFragmentDataFile); return result; } FragmentFilesMerger(ParsedStreamingCubeInfo parsedCubeInfo, File segmentFolder); FragmentsMergeResult merge(List<DataSegmentFragment> fragmentList); void cleanMergeDirectory(); }
FragmentFilesMerger { public FragmentsMergeResult merge(List<DataSegmentFragment> fragmentList) throws IOException { if (!mergeWorkingDirectory.exists()) { mergeWorkingDirectory.mkdirs(); } else { logger.info("clean the merge working dir:{}", mergeWorkingDirectory.getAbsolutePath()); FileUtils.cleanDirectory(mergeWorkingDirectory); } Collections.sort(fragmentList); FragmentId mergedFragmentId = new FragmentId(fragmentList.get(0).getFragmentId().getStartId(), fragmentList.get(fragmentList.size() - 1).getFragmentId().getEndId()); List<FragmentData> fragmentDataList = Lists.newArrayList(); Map<TblColRef, List<Dictionary<String>>> dimDictListMap = Maps.newHashMap(); Map<FragmentId, Map<TblColRef, Dictionary<String>>> fragmentDictionaryMaps = Maps.newHashMap(); List<Long> additionalCuboidsToMerge = null; long minMergedFragmentEventTime = Long.MAX_VALUE; long maxMergedFragmentEventTime = 0; long originNumOfRows = 0; for (DataSegmentFragment fragment : fragmentList) { FragmentData fragmentData = ColumnarStoreCache.getInstance().startReadFragmentData(fragment); FragmentMetaInfo fragmentMetaInfo = fragmentData.getFragmentMetaInfo(); long fragmentMinTime = fragmentMetaInfo.getMinEventTime(); long fragmentMaxTime = fragmentMetaInfo.getMaxEventTime(); originNumOfRows += fragmentMetaInfo.getOriginNumOfRows(); if (fragmentMinTime < minMergedFragmentEventTime) { minMergedFragmentEventTime = fragmentMinTime; } if (fragmentMaxTime > maxMergedFragmentEventTime) { maxMergedFragmentEventTime = fragmentMaxTime; } if (additionalCuboidsToMerge == null) { Map<String, CuboidMetaInfo> cuboidMetaInfoMap = fragmentMetaInfo.getCuboidMetaInfoMap(); if (cuboidMetaInfoMap != null) { additionalCuboidsToMerge = Lists.transform(Lists.newArrayList(cuboidMetaInfoMap.keySet()), new Function<String, Long>() { @Nullable @Override public Long apply(@Nullable String input) { return Long.valueOf(input); } }); } else { additionalCuboidsToMerge = Lists.newArrayList(); } } fragmentDataList.add(fragmentData); Map<TblColRef, Dictionary<String>> dictionaryMap = fragmentData .getDimensionDictionaries(parsedCubeInfo.dimensionsUseDictEncoding); fragmentDictionaryMaps.put(fragment.getFragmentId(), dictionaryMap); for (Entry<TblColRef, Dictionary<String>> dimDictEntry : dictionaryMap.entrySet()) { List<Dictionary<String>> dictionaryList = dimDictListMap.get(dimDictEntry.getKey()); if (dictionaryList == null) { dictionaryList = Lists.newArrayList(); dimDictListMap.put(dimDictEntry.getKey(), dictionaryList); } dictionaryList.add(dimDictEntry.getValue()); } } File mergedFragmentDataFile = new File(mergeWorkingDirectory, mergedFragmentId + Constants.DATA_FILE_SUFFIX); File mergedFragmentMetaFile = new File(mergeWorkingDirectory, mergedFragmentId + Constants.META_FILE_SUFFIX); try { FragmentMetaInfo mergedFragmentMeta = new FragmentMetaInfo(); CountingOutputStream fragmentDataOutput = new CountingOutputStream( new BufferedOutputStream(FileUtils.openOutputStream(mergedFragmentDataFile))); Map<TblColRef, Dictionary<String>> mergedDictMap = mergeAndPersistDictionaries(mergedFragmentMeta, dimDictListMap, fragmentDataOutput); logger.info("merge basic cuboid"); CuboidMetaInfo basicCuboidMeta = mergeAndPersistCuboidData(fragmentDataList, fragmentDictionaryMaps, mergedDictMap, fragmentDataOutput, parsedCubeInfo.basicCuboid.getId()); mergedFragmentMeta.setBasicCuboidMetaInfo(basicCuboidMeta); long totalRowCnt = basicCuboidMeta.getNumberOfRows(); Map<String, CuboidMetaInfo> cuboidMetaInfoMap = Maps.newHashMap(); for (Long cuboidId : additionalCuboidsToMerge) { logger.info("merge cuboid:{}", cuboidId); CuboidMetaInfo cuboidMeta = mergeAndPersistCuboidData(fragmentDataList, fragmentDictionaryMaps, mergedDictMap, fragmentDataOutput, cuboidId); cuboidMetaInfoMap.put(String.valueOf(cuboidId), cuboidMeta); totalRowCnt += cuboidMeta.getNumberOfRows(); } mergedFragmentMeta.setMaxEventTime(maxMergedFragmentEventTime); mergedFragmentMeta.setMinEventTime(minMergedFragmentEventTime); mergedFragmentMeta.setCuboidMetaInfoMap(cuboidMetaInfoMap); mergedFragmentMeta.setFragmentId(mergedFragmentId.toString()); mergedFragmentMeta.setNumberOfRows(totalRowCnt); mergedFragmentMeta.setOriginNumOfRows(originNumOfRows); fragmentDataOutput.flush(); fragmentDataOutput.close(); FileOutputStream metaOutputStream = FileUtils.openOutputStream(mergedFragmentMetaFile); JsonUtil.writeValueIndent(metaOutputStream, mergedFragmentMeta); metaOutputStream.flush(); metaOutputStream.close(); } finally { for (DataSegmentFragment fragment : fragmentList) { ColumnarStoreCache.getInstance().finishReadFragmentData(fragment); } } FragmentsMergeResult result = new FragmentsMergeResult(fragmentList, mergedFragmentId, mergedFragmentMetaFile, mergedFragmentDataFile); return result; } FragmentFilesMerger(ParsedStreamingCubeInfo parsedCubeInfo, File segmentFolder); FragmentsMergeResult merge(List<DataSegmentFragment> fragmentList); void cleanMergeDirectory();  }