@Test public void testConvert() throws Exception { MergeRollupTaskExecutor mergeRollupTaskExecutor = new MergeRollupTaskExecutor(); Map<String, String> configs = new HashMap<>(); configs.put(MinionConstants.MergeRollupTask.MERGE_TYPE_KEY, MergeType.CONCATENATE.toString()); configs.put(MinionConstants.MergeRollupTask.MERGED_SEGMENT_NAME_KEY, MERGED_SEGMENT_NAME); configs.put(MinionConstants.TABLE_NAME_KEY, "testTable_OFFLINE"); PinotTaskConfig pinotTaskConfig = new PinotTaskConfig(MinionConstants.MergeRollupTask.TASK_TYPE, configs); List<SegmentConversionResult> conversionResults = mergeRollupTaskExecutor.convert(pinotTaskConfig, _segmentIndexDirList, WORKING_DIR); Assert.assertEquals(conversionResults.size(), 1); Assert.assertEquals(conversionResults.get(0).getSegmentName(), MERGED_SEGMENT_NAME); File mergedSegment = conversionResults.get(0).getFile(); try (PinotSegmentRecordReader pinotSegmentRecordReader = new PinotSegmentRecordReader(mergedSegment)) { int numRecords = 0; GenericRow row = new GenericRow(); while (pinotSegmentRecordReader.hasNext()) { row = pinotSegmentRecordReader.next(row); numRecords++; } Assert.assertEquals(numRecords, NUM_SEGMENTS * NUM_ROWS); } }
@Override protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs, File workingDir) throws Exception { Map<String, String> configs = pinotTaskConfig.getConfigs(); String mergeTypeString = configs.get(MinionConstants.MergeRollupTask.MERGE_TYPE_KEY); Preconditions.checkNotNull(mergeTypeString, "MergeType cannot be null"); MergeType mergeType = MergeType.fromString(mergeTypeString); Preconditions.checkState(mergeType == MergeType.CONCATENATE, "Only 'CONCATENATE' mode is currently supported."); String mergedSegmentName = configs.get(MinionConstants.MergeRollupTask.MERGED_SEGMENT_NAME_KEY); String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); TableConfig tableConfig = getTableConfig(tableNameWithType); MergeRollupSegmentConverter rollupSegmentConverter = new MergeRollupSegmentConverter.Builder().setMergeType(mergeType).setTableName(tableNameWithType) .setSegmentName(mergedSegmentName).setInputIndexDirs(originalIndexDirs).setWorkingDir(workingDir) .setTableConfig(tableConfig).build(); List<File> resultFiles = rollupSegmentConverter.convert(); List<SegmentConversionResult> results = new ArrayList<>(); for (File file : resultFiles) { String outputSegmentName = file.getName(); results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName) .setTableNameWithType(tableNameWithType).build()); } return results; }
MergeRollupTaskExecutor extends BaseMultipleSegmentsConversionExecutor { @Override protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs, File workingDir) throws Exception { Map<String, String> configs = pinotTaskConfig.getConfigs(); String mergeTypeString = configs.get(MinionConstants.MergeRollupTask.MERGE_TYPE_KEY); Preconditions.checkNotNull(mergeTypeString, "MergeType cannot be null"); MergeType mergeType = MergeType.fromString(mergeTypeString); Preconditions.checkState(mergeType == MergeType.CONCATENATE, "Only 'CONCATENATE' mode is currently supported."); String mergedSegmentName = configs.get(MinionConstants.MergeRollupTask.MERGED_SEGMENT_NAME_KEY); String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); TableConfig tableConfig = getTableConfig(tableNameWithType); MergeRollupSegmentConverter rollupSegmentConverter = new MergeRollupSegmentConverter.Builder().setMergeType(mergeType).setTableName(tableNameWithType) .setSegmentName(mergedSegmentName).setInputIndexDirs(originalIndexDirs).setWorkingDir(workingDir) .setTableConfig(tableConfig).build(); List<File> resultFiles = rollupSegmentConverter.convert(); List<SegmentConversionResult> results = new ArrayList<>(); for (File file : resultFiles) { String outputSegmentName = file.getName(); results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName) .setTableNameWithType(tableNameWithType).build()); } return results; } }
MergeRollupTaskExecutor extends BaseMultipleSegmentsConversionExecutor { @Override protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs, File workingDir) throws Exception { Map<String, String> configs = pinotTaskConfig.getConfigs(); String mergeTypeString = configs.get(MinionConstants.MergeRollupTask.MERGE_TYPE_KEY); Preconditions.checkNotNull(mergeTypeString, "MergeType cannot be null"); MergeType mergeType = MergeType.fromString(mergeTypeString); Preconditions.checkState(mergeType == MergeType.CONCATENATE, "Only 'CONCATENATE' mode is currently supported."); String mergedSegmentName = configs.get(MinionConstants.MergeRollupTask.MERGED_SEGMENT_NAME_KEY); String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); TableConfig tableConfig = getTableConfig(tableNameWithType); MergeRollupSegmentConverter rollupSegmentConverter = new MergeRollupSegmentConverter.Builder().setMergeType(mergeType).setTableName(tableNameWithType) .setSegmentName(mergedSegmentName).setInputIndexDirs(originalIndexDirs).setWorkingDir(workingDir) .setTableConfig(tableConfig).build(); List<File> resultFiles = rollupSegmentConverter.convert(); List<SegmentConversionResult> results = new ArrayList<>(); for (File file : resultFiles) { String outputSegmentName = file.getName(); results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName) .setTableNameWithType(tableNameWithType).build()); } return results; }  }
MergeRollupTaskExecutor extends BaseMultipleSegmentsConversionExecutor { @Override protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs, File workingDir) throws Exception { Map<String, String> configs = pinotTaskConfig.getConfigs(); String mergeTypeString = configs.get(MinionConstants.MergeRollupTask.MERGE_TYPE_KEY); Preconditions.checkNotNull(mergeTypeString, "MergeType cannot be null"); MergeType mergeType = MergeType.fromString(mergeTypeString); Preconditions.checkState(mergeType == MergeType.CONCATENATE, "Only 'CONCATENATE' mode is currently supported."); String mergedSegmentName = configs.get(MinionConstants.MergeRollupTask.MERGED_SEGMENT_NAME_KEY); String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); TableConfig tableConfig = getTableConfig(tableNameWithType); MergeRollupSegmentConverter rollupSegmentConverter = new MergeRollupSegmentConverter.Builder().setMergeType(mergeType).setTableName(tableNameWithType) .setSegmentName(mergedSegmentName).setInputIndexDirs(originalIndexDirs).setWorkingDir(workingDir) .setTableConfig(tableConfig).build(); List<File> resultFiles = rollupSegmentConverter.convert(); List<SegmentConversionResult> results = new ArrayList<>(); for (File file : resultFiles) { String outputSegmentName = file.getName(); results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName) .setTableNameWithType(tableNameWithType).build()); } return results; }   }
MergeRollupTaskExecutor extends BaseMultipleSegmentsConversionExecutor { @Override protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs, File workingDir) throws Exception { Map<String, String> configs = pinotTaskConfig.getConfigs(); String mergeTypeString = configs.get(MinionConstants.MergeRollupTask.MERGE_TYPE_KEY); Preconditions.checkNotNull(mergeTypeString, "MergeType cannot be null"); MergeType mergeType = MergeType.fromString(mergeTypeString); Preconditions.checkState(mergeType == MergeType.CONCATENATE, "Only 'CONCATENATE' mode is currently supported."); String mergedSegmentName = configs.get(MinionConstants.MergeRollupTask.MERGED_SEGMENT_NAME_KEY); String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); TableConfig tableConfig = getTableConfig(tableNameWithType); MergeRollupSegmentConverter rollupSegmentConverter = new MergeRollupSegmentConverter.Builder().setMergeType(mergeType).setTableName(tableNameWithType) .setSegmentName(mergedSegmentName).setInputIndexDirs(originalIndexDirs).setWorkingDir(workingDir) .setTableConfig(tableConfig).build(); List<File> resultFiles = rollupSegmentConverter.convert(); List<SegmentConversionResult> results = new ArrayList<>(); for (File file : resultFiles) { String outputSegmentName = file.getName(); results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName) .setTableNameWithType(tableNameWithType).build()); } return results; }    }