@Test public void testPurgeSegment() throws Exception { SegmentPurger.RecordPurger recordPurger = row -> row.getValue(D1).equals(0); SegmentPurger.RecordModifier recordModifier = row -> { if (row.getValue(D2).equals(0)) { row.putValue(D2, Integer.MAX_VALUE); return true; } else { return false; } }; SegmentPurger segmentPurger = new SegmentPurger(_originalIndexDir, PURGED_SEGMENT_DIR, _tableConfig, recordPurger, recordModifier); File purgedIndexDir = segmentPurger.purgeSegment(); assertEquals(segmentPurger.getNumRecordsPurged(), _expectedNumRecordsPurged); assertEquals(segmentPurger.getNumRecordsModified(), _expectedNumRecordsModified); SegmentMetadataImpl purgedSegmentMetadata = new SegmentMetadataImpl(purgedIndexDir); SegmentMetadataImpl originalSegmentMetadata = new SegmentMetadataImpl(_originalIndexDir); assertNotEquals(purgedSegmentMetadata.getCrc(), originalSegmentMetadata.getCrc()); assertEquals(purgedSegmentMetadata.getIndexCreationTime(), originalSegmentMetadata.getIndexCreationTime()); try (PinotSegmentRecordReader pinotSegmentRecordReader = new PinotSegmentRecordReader(purgedIndexDir)) { int numRecordsRemaining = 0; int numRecordsModified = 0; GenericRow row = new GenericRow(); while (pinotSegmentRecordReader.hasNext()) { row = pinotSegmentRecordReader.next(row); assertNotEquals(row.getValue(D1), 0); assertNotEquals(row.getValue(D2), 0); numRecordsRemaining++; if (row.getValue(D2).equals(Integer.MAX_VALUE)) { numRecordsModified++; } } assertEquals(numRecordsRemaining, NUM_ROWS - _expectedNumRecordsPurged); assertEquals(numRecordsModified, _expectedNumRecordsModified); } try (SegmentDirectory segmentDirectory = SegmentDirectory .createFromLocalFS(purgedIndexDir, purgedSegmentMetadata, ReadMode.mmap); SegmentDirectory.Reader reader = segmentDirectory.createReader()) { assertTrue(reader.hasIndexFor(D1, ColumnIndexType.INVERTED_INDEX)); assertFalse(reader.hasIndexFor(D2, ColumnIndexType.INVERTED_INDEX)); } }
public File purgeSegment() throws Exception { SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(_indexDir); String segmentName = segmentMetadata.getName(); LOGGER.info("Start purging table: {}, segment: {}", _tableConfig.getTableName(), segmentName); try (PurgeRecordReader purgeRecordReader = new PurgeRecordReader()) { while (purgeRecordReader.hasNext()) { purgeRecordReader.next(); } if (_numRecordsModified == 0 && _numRecordsPurged == 0) { return null; } SegmentGeneratorConfig config = new SegmentGeneratorConfig(_tableConfig, purgeRecordReader.getSchema()); config.setOutDir(_workingDir.getPath()); config.setSegmentName(segmentName); config.setCreationTime(String.valueOf(segmentMetadata.getIndexCreationTime())); if (segmentMetadata.getTimeInterval() != null) { config.setTimeColumnName(_tableConfig.getValidationConfig().getTimeColumnName()); config.setStartTime(Long.toString(segmentMetadata.getStartTime())); config.setEndTime(Long.toString(segmentMetadata.getEndTime())); config.setSegmentTimeUnit(segmentMetadata.getTimeUnit()); } SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl(); purgeRecordReader.rewind(); driver.init(config, purgeRecordReader); driver.build(); } LOGGER.info("Finish purging table: {}, segment: {}, purged {} records, modified {} records", _tableConfig.getTableName(), segmentName, _numRecordsPurged, _numRecordsModified); return new File(_workingDir, segmentName); }
SegmentPurger { public File purgeSegment() throws Exception { SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(_indexDir); String segmentName = segmentMetadata.getName(); LOGGER.info("Start purging table: {}, segment: {}", _tableConfig.getTableName(), segmentName); try (PurgeRecordReader purgeRecordReader = new PurgeRecordReader()) { while (purgeRecordReader.hasNext()) { purgeRecordReader.next(); } if (_numRecordsModified == 0 && _numRecordsPurged == 0) { return null; } SegmentGeneratorConfig config = new SegmentGeneratorConfig(_tableConfig, purgeRecordReader.getSchema()); config.setOutDir(_workingDir.getPath()); config.setSegmentName(segmentName); config.setCreationTime(String.valueOf(segmentMetadata.getIndexCreationTime())); if (segmentMetadata.getTimeInterval() != null) { config.setTimeColumnName(_tableConfig.getValidationConfig().getTimeColumnName()); config.setStartTime(Long.toString(segmentMetadata.getStartTime())); config.setEndTime(Long.toString(segmentMetadata.getEndTime())); config.setSegmentTimeUnit(segmentMetadata.getTimeUnit()); } SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl(); purgeRecordReader.rewind(); driver.init(config, purgeRecordReader); driver.build(); } LOGGER.info("Finish purging table: {}, segment: {}, purged {} records, modified {} records", _tableConfig.getTableName(), segmentName, _numRecordsPurged, _numRecordsModified); return new File(_workingDir, segmentName); } }
SegmentPurger { public File purgeSegment() throws Exception { SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(_indexDir); String segmentName = segmentMetadata.getName(); LOGGER.info("Start purging table: {}, segment: {}", _tableConfig.getTableName(), segmentName); try (PurgeRecordReader purgeRecordReader = new PurgeRecordReader()) { while (purgeRecordReader.hasNext()) { purgeRecordReader.next(); } if (_numRecordsModified == 0 && _numRecordsPurged == 0) { return null; } SegmentGeneratorConfig config = new SegmentGeneratorConfig(_tableConfig, purgeRecordReader.getSchema()); config.setOutDir(_workingDir.getPath()); config.setSegmentName(segmentName); config.setCreationTime(String.valueOf(segmentMetadata.getIndexCreationTime())); if (segmentMetadata.getTimeInterval() != null) { config.setTimeColumnName(_tableConfig.getValidationConfig().getTimeColumnName()); config.setStartTime(Long.toString(segmentMetadata.getStartTime())); config.setEndTime(Long.toString(segmentMetadata.getEndTime())); config.setSegmentTimeUnit(segmentMetadata.getTimeUnit()); } SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl(); purgeRecordReader.rewind(); driver.init(config, purgeRecordReader); driver.build(); } LOGGER.info("Finish purging table: {}, segment: {}, purged {} records, modified {} records", _tableConfig.getTableName(), segmentName, _numRecordsPurged, _numRecordsModified); return new File(_workingDir, segmentName); } SegmentPurger(File indexDir, File workingDir, TableConfig tableConfig, @Nullable RecordPurger recordPurger,
      @Nullable RecordModifier recordModifier); }
SegmentPurger { public File purgeSegment() throws Exception { SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(_indexDir); String segmentName = segmentMetadata.getName(); LOGGER.info("Start purging table: {}, segment: {}", _tableConfig.getTableName(), segmentName); try (PurgeRecordReader purgeRecordReader = new PurgeRecordReader()) { while (purgeRecordReader.hasNext()) { purgeRecordReader.next(); } if (_numRecordsModified == 0 && _numRecordsPurged == 0) { return null; } SegmentGeneratorConfig config = new SegmentGeneratorConfig(_tableConfig, purgeRecordReader.getSchema()); config.setOutDir(_workingDir.getPath()); config.setSegmentName(segmentName); config.setCreationTime(String.valueOf(segmentMetadata.getIndexCreationTime())); if (segmentMetadata.getTimeInterval() != null) { config.setTimeColumnName(_tableConfig.getValidationConfig().getTimeColumnName()); config.setStartTime(Long.toString(segmentMetadata.getStartTime())); config.setEndTime(Long.toString(segmentMetadata.getEndTime())); config.setSegmentTimeUnit(segmentMetadata.getTimeUnit()); } SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl(); purgeRecordReader.rewind(); driver.init(config, purgeRecordReader); driver.build(); } LOGGER.info("Finish purging table: {}, segment: {}, purged {} records, modified {} records", _tableConfig.getTableName(), segmentName, _numRecordsPurged, _numRecordsModified); return new File(_workingDir, segmentName); } SegmentPurger(File indexDir, File workingDir, TableConfig tableConfig, @Nullable RecordPurger recordPurger,
      @Nullable RecordModifier recordModifier); File purgeSegment(); RecordPurger getRecordPurger(); RecordModifier getRecordModifier(); int getNumRecordsPurged(); int getNumRecordsModified(); }
SegmentPurger { public File purgeSegment() throws Exception { SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(_indexDir); String segmentName = segmentMetadata.getName(); LOGGER.info("Start purging table: {}, segment: {}", _tableConfig.getTableName(), segmentName); try (PurgeRecordReader purgeRecordReader = new PurgeRecordReader()) { while (purgeRecordReader.hasNext()) { purgeRecordReader.next(); } if (_numRecordsModified == 0 && _numRecordsPurged == 0) { return null; } SegmentGeneratorConfig config = new SegmentGeneratorConfig(_tableConfig, purgeRecordReader.getSchema()); config.setOutDir(_workingDir.getPath()); config.setSegmentName(segmentName); config.setCreationTime(String.valueOf(segmentMetadata.getIndexCreationTime())); if (segmentMetadata.getTimeInterval() != null) { config.setTimeColumnName(_tableConfig.getValidationConfig().getTimeColumnName()); config.setStartTime(Long.toString(segmentMetadata.getStartTime())); config.setEndTime(Long.toString(segmentMetadata.getEndTime())); config.setSegmentTimeUnit(segmentMetadata.getTimeUnit()); } SegmentIndexCreationDriverImpl driver = new SegmentIndexCreationDriverImpl(); purgeRecordReader.rewind(); driver.init(config, purgeRecordReader); driver.build(); } LOGGER.info("Finish purging table: {}, segment: {}, purged {} records, modified {} records", _tableConfig.getTableName(), segmentName, _numRecordsPurged, _numRecordsModified); return new File(_workingDir, segmentName); } SegmentPurger(File indexDir, File workingDir, TableConfig tableConfig, @Nullable RecordPurger recordPurger,
      @Nullable RecordModifier recordModifier); File purgeSegment(); RecordPurger getRecordPurger(); RecordModifier getRecordModifier(); int getNumRecordsPurged(); int getNumRecordsModified();  }