Page History

public interface DeleteIndex {

    void delete(long position);

    boolean isDeleted(long position);

    boolean isEmpty();

    byte[] serializeToBytes();

    DeleteIndex deserializeFromBytes(byte[] bytes);
}

public class DeleteMapIndexFile {  
   
    public long fileSize(String fileName);
     
    public Map<String, long[]> readDeleteIndexBytesOffsets(String fileName);
    
    public Map<String, DeleteIndex> readAllDeleteIndex(String fileName, Map<String, long[]> deleteIndexBytesOffsets);
    
    public DeleteIndex readDeleteIndex(String fileName, long[] deleteIndexBytesOffset);

    public String write(Map<String, DeleteIndex> input);

    public void delete(String fileName);
}

public class ApplyDeleteIndexReader implements RecordReader<KeyValue> {

   public ApplyDeleteIndexReader(RecordReader<KeyValue> reader, DeleteIndex deleteIndex) {
        this.reader = reader;
        this.deleteIndex = deleteIndex;
    }

    @Nullable
    @Override
    public RecordIterator<KeyValue> readBatch() throws IOException {
        RecordIterator<KeyValue> batch = reader.readBatch();
        if (batch == null) {
            return null;
        }
        return new RecordIterator<KeyValue>() {
            @Override
            public KeyValue next() throws IOException {
                while (true) {
                    KeyValue kv = batch.next();
                    if (kv == null) {
                        return null;
                    }
                    if (!deleteIndex.isDeleted(kv.position())) {
                        return kv;
                    }
                }
            }
        };
    }
  ...
}