@Test public void testConvert() throws Exception { PurgeTaskExecutor purgeTaskExecutor = new PurgeTaskExecutor(); PinotTaskConfig pinotTaskConfig = new PinotTaskConfig(MinionConstants.PurgeTask.TASK_TYPE, Collections .singletonMap(MinionConstants.TABLE_NAME_KEY, TableNameBuilder.OFFLINE.tableNameWithType(TABLE_NAME))); File purgedIndexDir = purgeTaskExecutor.convert(pinotTaskConfig, _originalIndexDir, PURGED_SEGMENT_DIR).getFile(); try (PinotSegmentRecordReader pinotSegmentRecordReader = new PinotSegmentRecordReader(purgedIndexDir)) { int numRecordsRemaining = 0; int numRecordsModified = 0; GenericRow row = new GenericRow(); while (pinotSegmentRecordReader.hasNext()) { row = pinotSegmentRecordReader.next(row); numRecordsRemaining++; if (row.getValue(D1).equals(Integer.MAX_VALUE)) { numRecordsModified++; } } assertEquals(numRecordsRemaining, NUM_ROWS - 1); assertEquals(numRecordsModified, NUM_ROWS - 1); } }
@Override protected SegmentConversionResult convert(PinotTaskConfig pinotTaskConfig, File indexDir, File workingDir) throws Exception { Map<String, String> configs = pinotTaskConfig.getConfigs(); String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); String rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType); TableConfig tableConfig = getTableConfig(tableNameWithType); SegmentPurger.RecordPurgerFactory recordPurgerFactory = MINION_CONTEXT.getRecordPurgerFactory(); SegmentPurger.RecordPurger recordPurger = recordPurgerFactory != null ? recordPurgerFactory.getRecordPurger(rawTableName) : null; SegmentPurger.RecordModifierFactory recordModifierFactory = MINION_CONTEXT.getRecordModifierFactory(); SegmentPurger.RecordModifier recordModifier = recordModifierFactory != null ? recordModifierFactory.getRecordModifier(rawTableName) : null; SegmentPurger segmentPurger = new SegmentPurger(indexDir, workingDir, tableConfig, recordPurger, recordModifier); File purgedSegmentFile = segmentPurger.purgeSegment(); if (purgedSegmentFile == null) { purgedSegmentFile = indexDir; } return new SegmentConversionResult.Builder().setFile(purgedSegmentFile).setTableNameWithType(tableNameWithType) .setSegmentName(configs.get(MinionConstants.SEGMENT_NAME_KEY)) .setCustomProperty(RECORD_PURGER_KEY, segmentPurger.getRecordPurger()) .setCustomProperty(RECORD_MODIFIER_KEY, segmentPurger.getRecordModifier()) .setCustomProperty(NUM_RECORDS_PURGED_KEY, segmentPurger.getNumRecordsPurged()) .setCustomProperty(NUM_RECORDS_MODIFIED_KEY, segmentPurger.getNumRecordsModified()).build(); }
PurgeTaskExecutor extends BaseSingleSegmentConversionExecutor { @Override protected SegmentConversionResult convert(PinotTaskConfig pinotTaskConfig, File indexDir, File workingDir) throws Exception { Map<String, String> configs = pinotTaskConfig.getConfigs(); String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); String rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType); TableConfig tableConfig = getTableConfig(tableNameWithType); SegmentPurger.RecordPurgerFactory recordPurgerFactory = MINION_CONTEXT.getRecordPurgerFactory(); SegmentPurger.RecordPurger recordPurger = recordPurgerFactory != null ? recordPurgerFactory.getRecordPurger(rawTableName) : null; SegmentPurger.RecordModifierFactory recordModifierFactory = MINION_CONTEXT.getRecordModifierFactory(); SegmentPurger.RecordModifier recordModifier = recordModifierFactory != null ? recordModifierFactory.getRecordModifier(rawTableName) : null; SegmentPurger segmentPurger = new SegmentPurger(indexDir, workingDir, tableConfig, recordPurger, recordModifier); File purgedSegmentFile = segmentPurger.purgeSegment(); if (purgedSegmentFile == null) { purgedSegmentFile = indexDir; } return new SegmentConversionResult.Builder().setFile(purgedSegmentFile).setTableNameWithType(tableNameWithType) .setSegmentName(configs.get(MinionConstants.SEGMENT_NAME_KEY)) .setCustomProperty(RECORD_PURGER_KEY, segmentPurger.getRecordPurger()) .setCustomProperty(RECORD_MODIFIER_KEY, segmentPurger.getRecordModifier()) .setCustomProperty(NUM_RECORDS_PURGED_KEY, segmentPurger.getNumRecordsPurged()) .setCustomProperty(NUM_RECORDS_MODIFIED_KEY, segmentPurger.getNumRecordsModified()).build(); } }
PurgeTaskExecutor extends BaseSingleSegmentConversionExecutor { @Override protected SegmentConversionResult convert(PinotTaskConfig pinotTaskConfig, File indexDir, File workingDir) throws Exception { Map<String, String> configs = pinotTaskConfig.getConfigs(); String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); String rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType); TableConfig tableConfig = getTableConfig(tableNameWithType); SegmentPurger.RecordPurgerFactory recordPurgerFactory = MINION_CONTEXT.getRecordPurgerFactory(); SegmentPurger.RecordPurger recordPurger = recordPurgerFactory != null ? recordPurgerFactory.getRecordPurger(rawTableName) : null; SegmentPurger.RecordModifierFactory recordModifierFactory = MINION_CONTEXT.getRecordModifierFactory(); SegmentPurger.RecordModifier recordModifier = recordModifierFactory != null ? recordModifierFactory.getRecordModifier(rawTableName) : null; SegmentPurger segmentPurger = new SegmentPurger(indexDir, workingDir, tableConfig, recordPurger, recordModifier); File purgedSegmentFile = segmentPurger.purgeSegment(); if (purgedSegmentFile == null) { purgedSegmentFile = indexDir; } return new SegmentConversionResult.Builder().setFile(purgedSegmentFile).setTableNameWithType(tableNameWithType) .setSegmentName(configs.get(MinionConstants.SEGMENT_NAME_KEY)) .setCustomProperty(RECORD_PURGER_KEY, segmentPurger.getRecordPurger()) .setCustomProperty(RECORD_MODIFIER_KEY, segmentPurger.getRecordModifier()) .setCustomProperty(NUM_RECORDS_PURGED_KEY, segmentPurger.getNumRecordsPurged()) .setCustomProperty(NUM_RECORDS_MODIFIED_KEY, segmentPurger.getNumRecordsModified()).build(); }  }
PurgeTaskExecutor extends BaseSingleSegmentConversionExecutor { @Override protected SegmentConversionResult convert(PinotTaskConfig pinotTaskConfig, File indexDir, File workingDir) throws Exception { Map<String, String> configs = pinotTaskConfig.getConfigs(); String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); String rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType); TableConfig tableConfig = getTableConfig(tableNameWithType); SegmentPurger.RecordPurgerFactory recordPurgerFactory = MINION_CONTEXT.getRecordPurgerFactory(); SegmentPurger.RecordPurger recordPurger = recordPurgerFactory != null ? recordPurgerFactory.getRecordPurger(rawTableName) : null; SegmentPurger.RecordModifierFactory recordModifierFactory = MINION_CONTEXT.getRecordModifierFactory(); SegmentPurger.RecordModifier recordModifier = recordModifierFactory != null ? recordModifierFactory.getRecordModifier(rawTableName) : null; SegmentPurger segmentPurger = new SegmentPurger(indexDir, workingDir, tableConfig, recordPurger, recordModifier); File purgedSegmentFile = segmentPurger.purgeSegment(); if (purgedSegmentFile == null) { purgedSegmentFile = indexDir; } return new SegmentConversionResult.Builder().setFile(purgedSegmentFile).setTableNameWithType(tableNameWithType) .setSegmentName(configs.get(MinionConstants.SEGMENT_NAME_KEY)) .setCustomProperty(RECORD_PURGER_KEY, segmentPurger.getRecordPurger()) .setCustomProperty(RECORD_MODIFIER_KEY, segmentPurger.getRecordModifier()) .setCustomProperty(NUM_RECORDS_PURGED_KEY, segmentPurger.getNumRecordsPurged()) .setCustomProperty(NUM_RECORDS_MODIFIED_KEY, segmentPurger.getNumRecordsModified()).build(); }   }
PurgeTaskExecutor extends BaseSingleSegmentConversionExecutor { @Override protected SegmentConversionResult convert(PinotTaskConfig pinotTaskConfig, File indexDir, File workingDir) throws Exception { Map<String, String> configs = pinotTaskConfig.getConfigs(); String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); String rawTableName = TableNameBuilder.extractRawTableName(tableNameWithType); TableConfig tableConfig = getTableConfig(tableNameWithType); SegmentPurger.RecordPurgerFactory recordPurgerFactory = MINION_CONTEXT.getRecordPurgerFactory(); SegmentPurger.RecordPurger recordPurger = recordPurgerFactory != null ? recordPurgerFactory.getRecordPurger(rawTableName) : null; SegmentPurger.RecordModifierFactory recordModifierFactory = MINION_CONTEXT.getRecordModifierFactory(); SegmentPurger.RecordModifier recordModifier = recordModifierFactory != null ? recordModifierFactory.getRecordModifier(rawTableName) : null; SegmentPurger segmentPurger = new SegmentPurger(indexDir, workingDir, tableConfig, recordPurger, recordModifier); File purgedSegmentFile = segmentPurger.purgeSegment(); if (purgedSegmentFile == null) { purgedSegmentFile = indexDir; } return new SegmentConversionResult.Builder().setFile(purgedSegmentFile).setTableNameWithType(tableNameWithType) .setSegmentName(configs.get(MinionConstants.SEGMENT_NAME_KEY)) .setCustomProperty(RECORD_PURGER_KEY, segmentPurger.getRecordPurger()) .setCustomProperty(RECORD_MODIFIER_KEY, segmentPurger.getRecordModifier()) .setCustomProperty(NUM_RECORDS_PURGED_KEY, segmentPurger.getNumRecordsPurged()) .setCustomProperty(NUM_RECORDS_MODIFIED_KEY, segmentPurger.getNumRecordsModified()).build(); }   static final String RECORD_PURGER_KEY; static final String RECORD_MODIFIER_KEY; static final String NUM_RECORDS_PURGED_KEY; static final String NUM_RECORDS_MODIFIED_KEY; }