public class Tuple implements Comparable<Tuple> {

    public static Serializer<Tuple> SERIALIZER = new Serializer<Tuple>() {
        @Override
        public int size(Tuple tuple) {
            int size = 0;

            //Key
            size += Sizes.INT_SIZE;
            size += tuple.key().size();
            size += Sizes.LONG_SIZE;

            //Value
            size += Sizes.INT_SIZE;
            size += tuple.value().size();

            return size;
        }

        @Override
        public void serialize(Tuple tuple, ByteBuffer recordBuffer) {
            //Key
            recordBuffer.putInt(tuple.key.size());
            recordBuffer.put(tuple.key.data());
            recordBuffer.putLong(tuple.key.snapshotId());
            tuple.key().data().rewind();

            //Value
            recordBuffer.putInt(tuple.value.size());
            recordBuffer.put(tuple.value().data());
            tuple.value().data().rewind();

            recordBuffer.rewind();
        }

        @Override
        public Tuple deserialize(ByteBuffer recordBuffer) {
            //Key
            int keySize = recordBuffer.getInt();
            ByteBuffer keyBuffer = ByteBuffer.allocate(keySize);
            recordBuffer.get(keyBuffer.array());
            long snapshotId = recordBuffer.getLong();
            Key key = new Key(keyBuffer, snapshotId);

            //Value
            int valueSize = recordBuffer.getInt();
            ByteBuffer valueBuffer = ByteBuffer.allocate(valueSize);
            recordBuffer.get(valueBuffer.array());
            Value value = new Value(valueBuffer);

            return new Tuple(key, value);
        }
    };

    private final Key key;
    private final Value value;

    public Tuple(Key key, Value value) {
        this.key = key;
        this.value = value;
    }

    public Key key() {
        return key;
    }

    public Value value() {
        return value;
    }

    public int size() {
        return key.size() + value().size();
    }

    public void rewind() {
        key.data().rewind();
        value.data().rewind();
    }

    @Override
    public int compareTo(Tuple o) {
        return key.compareTo(o.key);
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;

        Tuple tuple = (Tuple) o;

        if (key != null ? !key.equals(tuple.key) : tuple.key != null) return false;
        if (value != null ? !value.equals(tuple.value) : tuple.value != null) return false;

        return true;
    }

    @Override
    public int hashCode() {
        int result = key != null ? key.hashCode() : 0;
        result = 31 * result + (value != null ? value.hashCode() : 0);
        return result;
    }

    @Override
    public String toString() {
        return "Tuple{" +
                "key=" + key +
                ", value=" + value +
                '}';
    }
} public class IndexBlock implements Iterable<IndexRecord>, Offheap {

    public static class Cache {

        private final TableBlockCache<IndexBlock> cache;

        public Cache(long maxSize, Metrics metrics) {
            cache = new TableBlockCache<IndexBlock>(maxSize, new Weigher<IndexBlock>() {
                @Override
                public int weightOf(IndexBlock indexRecord) {
                    return indexRecord.memory().size();
                }
            });

            metrics.gauge("cache.indexBlock.entrySize", new Gauge<Long>() {
                @Override
                public Long getValue() {
                    return cache.totalEntrySize();
                }
            });

            metrics.gauge("cache.indexBlock.utilizationRate", new Gauge<Double>() {
                @Override
                public Double getValue() {
                    return cache.utilizationRate();
                }
            });
        }

        public IndexBlock get(long tableId, long offset) {
            return cache.get(tableId, offset);
        }

        public void put(long tableId, long offset, IndexBlock tupleBlock) {
            cache.put(tableId, offset, tupleBlock);
        }

        public void invalidate(long tableId) {
            cache.invalidate(tableId);
        }

        public void clear() {
            cache.clear();
        }
    }

    public static class Builder {

        private final SortedByteMap.Builder byteMapBuilder = new SortedByteMap.Builder();

        private int size;

        public void addRecord(IndexRecord indexRecord) {
            byteMapBuilder.add(indexRecord.startKey(), new Value(indexRecordValue(indexRecord)));
            size += indexRecord.size();
        }

        public int size() {
            return size;
        }

        public IndexBlock build() {
            return new IndexBlock(byteMapBuilder.build());
        }

        private ByteBuffer indexRecordValue(IndexRecord indexRecord) {
            ByteBuffer contentsBuffer = ByteBuffer.allocate(indexRecord.contentsSize());
            contentsBuffer.putLong(indexRecord.blockOffset());
            contentsBuffer.putInt(indexRecord.blockSize());
            contentsBuffer.put(indexRecord.isLeaf() ? (byte) 1 : (byte) 0);
            contentsBuffer.rewind();
            return contentsBuffer;
        }
    }

    private class RecordIterator implements Iterator<IndexRecord> {

        private final Iterator<SortedByteMap.Entry> entryIterator;

        private RecordIterator(Iterator<SortedByteMap.Entry> entryIterator) {
            this.entryIterator = entryIterator;
        }

        @Override
        public boolean hasNext() {
            return entryIterator.hasNext();
        }

        @Override
        public IndexRecord next() {
            SortedByteMap.Entry nextEntry = entryIterator.next();
            return deserialize(nextEntry);
        }

        @Override
        public void remove() {
            entryIterator.remove();
        }
    }

    private final SortedByteMap sortedByteMap;

    public IndexBlock(SortedByteMap sortedByteMap) {
        this.sortedByteMap = sortedByteMap;
    }

    public IndexRecord startRecord() {
        return deserialize(0);
    }

    public IndexRecord get(Key key) {
        int closestIndex = sortedByteMap.floorIndex(key);

        if (closestIndex < 0) {
            return null;
        }

        if (closestIndex >= sortedByteMap.entryCount()) {
            closestIndex = sortedByteMap.entryCount() - 1;
        }

        return deserialize(closestIndex);
    }

    @Override
    public MemoryPointer memory() {
        return sortedByteMap.memory();
    }

    public Iterator<IndexRecord> ascendingIterator() {
        return new RecordIterator(sortedByteMap.ascendingIterator());
    }

    public Iterator<IndexRecord> ascendingIterator(Key key) {
        return new RecordIterator(sortedByteMap.ascendingIterator(key));
    }

    public Iterator<IndexRecord> descendingIterator() {
        return new RecordIterator(sortedByteMap.descendingIterator());
    }

    public Iterator<IndexRecord> descendingIterator(Key key) {
        return new RecordIterator(sortedByteMap.descendingIterator(key));
    }

    @Override
    public Iterator<IndexRecord> iterator() {
        return new RecordIterator(sortedByteMap.ascendingIterator());
    }

    @Override
    public String toString() {
        List<IndexRecord> records = new ArrayList<IndexRecord>();

        for (IndexRecord indexRecord : this) {
            records.add(indexRecord);
        }

        return "IndexBlock{records=" + records + "}";
    }

    private IndexRecord deserialize(int index) {
        SortedByteMap.Entry entry = sortedByteMap.get(index);
        return deserialize(entry);
    }

    private IndexRecord deserialize(SortedByteMap.Entry entry) {
        ByteBuffer entryValueBuffer = entry.value().data();
        long blockOffset = entryValueBuffer.getLong(0);
        int blockSize = entryValueBuffer.getInt(Sizes.LONG_SIZE);
        boolean isLeaf = entryValueBuffer.get(Sizes.LONG_SIZE + Sizes.INT_SIZE) == (byte) 1;
        return new IndexRecord(entry.key(), blockOffset, blockSize, isLeaf);
    }
} public class Metrics {

    private static final Logger logger = LoggerFactory.getLogger(Metrics.class);
    private static final String METRIC_PREFIX = "heftydb.";

    private final Map<String, CacheHitGauge> gaugeCache = new ConcurrentHashMap<String, CacheHitGauge>();
    private final MetricRegistry metrics = new MetricRegistry();
    private final Slf4jReporter reporter;
    private final JmxReporter jmxReporter;

    public Metrics(Config config) {
        this.reporter = Slf4jReporter.forRegistry(metrics).outputTo(logger).convertDurationsTo(TimeUnit.MILLISECONDS)
                .convertRatesTo(TimeUnit.SECONDS).build();

        reporter.start(1, TimeUnit.MINUTES);

        this.jmxReporter = JmxReporter.forRegistry(metrics).inDomain("HeftyDB @ " + config.tableDirectory().toString
                ()).convertDurationsTo(TimeUnit.MILLISECONDS).convertRatesTo(TimeUnit.SECONDS).build();

        jmxReporter.start();

        initMetrics();
    }

    public void close() {
        jmxReporter.close();
    }

    public void logMetrics() {
        reporter.report();
    }

    public Counter counter(String name) {
        return metrics.counter(metricName(name));
    }

    public Meter meter(String name) {
        return metrics.meter(metricName(name));
    }

    public CacheHitGauge hitGauge(String name) {
        CacheHitGauge cacheHitGauge = gaugeCache.get(metricName(name));

        if (cacheHitGauge == null) {
            cacheHitGauge = (CacheHitGauge) metrics.getGauges().get(metricName(name));
            gaugeCache.put(metricName(name), cacheHitGauge);
        }

        return cacheHitGauge;
    }

    public void gauge(String name, Gauge<?> gauge) {
        metrics.register(metricName(name), gauge);
    }

    public Histogram histogram(String name) {
        return metrics.histogram(metricName(name));
    }

    public Timer timer(String name) {
        return metrics.timer(metricName(name));
    }

    private void initMetrics() {
        //Main DB Metrics
        metrics.register(metricName("write"), new Timer(new UniformReservoir()));
        metrics.register(metricName("write.rate"), new Meter());
        metrics.register(metricName("read"), new Timer(new UniformReservoir()));
        metrics.register(metricName("read.rate"), new Meter());
        metrics.register(metricName("scan"), new Timer(new UniformReservoir()));
        metrics.register(metricName("scan.rate"), new Meter());

        //Write
        metrics.register(metricName("write.concurrentMemoryTableSerializers"), new Histogram(new UniformReservoir()));
        metrics.register(metricName("write.memoryTableSerialize"), new Timer(new UniformReservoir()));

        //Read
        metrics.register(metricName("read.tablesConsulted"), new Histogram(new UniformReservoir()));
        metrics.register(metricName("read.bloomFilterFalsePositiveRate"), new CacheHitGauge());
        metrics.register(metricName("read.recordNotFoundRate"), new CacheHitGauge());

        //FileTable
        metrics.register(metricName("table.cacheHitRate"), new CacheHitGauge());

        //Index
        metrics.register(metricName("index.searchLevels"), new Histogram(new UniformReservoir()));
        metrics.register(metricName("index.cacheHitRate"), new CacheHitGauge());

        //Compactor
        metrics.register(metricName("compactor.concurrentTasks"), new Histogram(new UniformReservoir()));
        metrics.register(metricName("compactor.taskExecution"), new Timer(new UniformReservoir()));
    }

    private static String metricName(String name) {
        return METRIC_PREFIX + name;
    }
} public class Caches {

    private final TupleBlock.Cache recordBlockCache;
    private final IndexBlock.Cache indexBlockCache;

    public Caches(TupleBlock.Cache recordBlockCache, IndexBlock.Cache indexBlockCache) {
        this.recordBlockCache = recordBlockCache;
        this.indexBlockCache = indexBlockCache;
    }

    public TupleBlock.Cache recordBlockCache() {
        return recordBlockCache;
    }

    public IndexBlock.Cache indexBlockCache() {
        return indexBlockCache;
    }
} public class Paths {

    private static final String TABLE_EXT = ".table";
    private static final String LOG_EXT = ".log";
    private static final String INDEX_EXT = ".index";
    private static final String FILTER_EXT = ".filter";
    private static final String TEMP_EXT = ".temp";

    private final Path logDirectory;
    private final Path tableDirectory;

    public Paths(Path tableDirectory, Path logDirectory) {
        this.tableDirectory = tableDirectory;
        this.logDirectory = logDirectory;
    }

    public Path tablePath(long tableId) {
        return tableDirectory.resolve(tableId + TABLE_EXT);
    }

    public Path indexPath(long tableId) {
        return tableDirectory.resolve(tableId + INDEX_EXT);
    }

    public Path filterPath(long tableId) {
        return tableDirectory.resolve(tableId + FILTER_EXT);
    }

    public Path logPath(long tableId) {
        return tableDirectory.resolve(tableId + LOG_EXT);
    }

    public Path tempPath(long tableId) {
        return tableDirectory.resolve(tableId + TEMP_EXT);
    }

    public SortedSet<Long> tableFileIds() throws IOException {
        return fileIds(tableFilePaths());
    }

    public SortedSet<Long> tempTableFileIds() throws IOException {
        return fileIds(tempFilePaths());
    }

    public SortedSet<Long> logFileIds() throws IOException {
        return fileIds(logFilePaths());
    }

    public SortedSet<Path> tableFilePaths() throws IOException {
        return filePaths(tableDirectory, TABLE_EXT);
    }

    public SortedSet<Path> logFilePaths() throws IOException {
        return filePaths(tableDirectory, LOG_EXT);
    }

    public SortedSet<Path> tempFilePaths() throws IOException {
        return filePaths(tableDirectory, TEMP_EXT);
    }

    private SortedSet<Path> filePaths(Path directory, String extension) throws IOException {
        SortedSet<Path> sortedFilePaths = new TreeSet<Path>();
        DirectoryStream<Path> filePaths = Files.newDirectoryStream(directory, "*" + extension);

        for (Path filePath : filePaths) {
            sortedFilePaths.add(filePath);
        }

        filePaths.close();

        return sortedFilePaths;
    }

    private SortedSet<Long> fileIds(SortedSet<Path> paths) {
        SortedSet<Long> sortedFileIds = new TreeSet<Long>();

        for (Path path : paths) {
            sortedFileIds.add(tableId(path));
        }

        return sortedFileIds;
    }

    private static long tableId(Path path) {
        String fileName = path.getFileName().toString();
        String id = fileName.split("\\.")[0];
        return Long.parseLong(id);
    }
} public interface MutableTable extends Table {

    public void put(Tuple tuple);
} public interface Table extends Iterable<Tuple>, Comparable<Table> {

    public long id();

    public boolean mightContain(Key key);

    public Tuple get(Key key);

    public CloseableIterator<Tuple> ascendingIterator(long snapshotId);

    public CloseableIterator<Tuple> descendingIterator(long snapshotId);

    public CloseableIterator<Tuple> ascendingIterator(Key key, long snapshotId);

    public CloseableIterator<Tuple> descendingIterator(Key key, long snapshotId);

    public long tupleCount();

    public long size();

    public int level();

    public long maxSnapshotId();

    public void close();

    public boolean isPersistent();
} public class FileTable implements Table {

    private class AscendingBlockIterator implements Iterator<TupleBlock> {

        private final long maxOffset;
        private long fileOffset = 0;

        public AscendingBlockIterator(long startOffset) {
            this.fileOffset = startOffset;
            this.maxOffset = fileSize - TableTrailer.SIZE - Sizes.INT_SIZE;
        }

        @Override
        public boolean hasNext() {
            return fileOffset < maxOffset;
        }

        @Override
        public TupleBlock next() {
            if (!hasNext()) {
                throw new NoSuchElementException();
            }

            try {
                int nextBlockSize = tableFile.readInt(fileOffset);
                long nextBlockOffset = fileOffset + Sizes.INT_SIZE;

                fileOffset += Sizes.INT_SIZE;
                fileOffset += nextBlockSize;
                fileOffset += Sizes.INT_SIZE;

                return readTupleBlock(nextBlockOffset, nextBlockSize);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        public void remove() {
            throw new UnsupportedOperationException();
        }
    }

    private class DescendingBlockIterator implements Iterator<TupleBlock> {

        private long fileOffset;

        public DescendingBlockIterator(long startOffset) {
            this.fileOffset = startOffset;
        }

        @Override
        public boolean hasNext() {
            return fileOffset >= 0;
        }

        @Override
        public TupleBlock next() {
            if (!hasNext()) {
                throw new NoSuchElementException();
            }

            try {
                int nextBlockSize = tableFile.readInt(fileOffset);
                long nextBlockOffset = fileOffset - nextBlockSize;

                fileOffset -= Sizes.INT_SIZE;
                fileOffset -= nextBlockSize;
                fileOffset -= Sizes.INT_SIZE;

                return readTupleBlock(nextBlockOffset, nextBlockSize);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        public void remove() {
            throw new UnsupportedOperationException();
        }
    }

    private class AscendingIterator implements CloseableIterator<Tuple> {

        protected final Iterator<TupleBlock> recordBlockIterator;
        protected Iterator<Tuple> recordIterator;
        protected TupleBlock tupleBlock;

        private AscendingIterator(Iterator<TupleBlock> recordBlockIterator, Iterator<Tuple> startIterator,
                                  TupleBlock startTupleBlock) {
            this.recordBlockIterator = recordBlockIterator;
            this.recordIterator = startIterator;
            this.tupleBlock = startTupleBlock;
        }

        private AscendingIterator(Iterator<TupleBlock> recordBlockIterator) {
            this.recordBlockIterator = recordBlockIterator;
        }

        @Override
        public boolean hasNext() {
            try {
                if (recordIterator == null || !recordIterator.hasNext()) {
                    if (!nextRecordBlock()) {
                        return false;
                    }
                }

                if (recordIterator == null || !recordIterator.hasNext()) {
                    return false;
                }

                return true;
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }

        @Override
        public Tuple next() {
            if (recordIterator == null || !recordIterator.hasNext()) {
                if (!hasNext()) {
                    throw new NoSuchElementException();
                }
            }

            return recordIterator.next();
        }

        @Override
        public void remove() {
            throw new UnsupportedOperationException();
        }

        @Override
        public void close() throws IOException {
            if (tupleBlock != null && !tupleBlock.memory().isFree()) {
                tupleBlock.memory().free();
            }
        }

        protected boolean nextRecordBlock() throws IOException {
            if (tupleBlock != null) {
                tupleBlock.memory().free();
            }

            if (!recordBlockIterator.hasNext()) {
                return false;
            }

            tupleBlock = recordBlockIterator.next();
            recordIterator = tupleBlock.ascendingIterator();

            return true;
        }
    }

    private class DescendingIterator extends AscendingIterator {

        private DescendingIterator(Iterator<TupleBlock> recordBlockIterator, Iterator<Tuple> startIterator,
                                   TupleBlock startTupleBlock) {
            super(recordBlockIterator, startIterator, startTupleBlock);
        }

        private DescendingIterator(Iterator<TupleBlock> recordBlockIterator) {
            super(recordBlockIterator);
        }

        @Override
        protected boolean nextRecordBlock() throws IOException {
            if (tupleBlock != null) {
                tupleBlock.memory().free();
            }

            if (!recordBlockIterator.hasNext()) {
                return false;
            }

            tupleBlock = recordBlockIterator.next();
            recordIterator = tupleBlock.descendingIterator();

            return true;
        }
    }

    private final long tableId;
    private final long fileSize;
    private final Index index;
    private final TableBloomFilter tableBloomFilter;
    private final TableTrailer trailer;
    private final TupleBlock.Cache recordCache;
    private final ImmutableFile tableFile;
    private final Metrics metrics;

    private final CacheHitGauge tableCacheHitRate;

    private FileTable(long tableId, Index index, TableBloomFilter tableBloomFilter, ImmutableFile tableFile,
                      TableTrailer trailer, TupleBlock.Cache recordCache, Metrics metrics) throws IOException {
        this.tableId = tableId;
        this.recordCache = recordCache;
        this.index = index;
        this.tableBloomFilter = tableBloomFilter;
        this.tableFile = tableFile;
        this.trailer = trailer;
        this.metrics = metrics;
        this.fileSize = tableFile.size();

        this.tableCacheHitRate = metrics.hitGauge("table.cacheHitRate");
    }

    @Override
    public boolean mightContain(Key key) {
        return tableBloomFilter.mightContain(key);
    }

    @Override
    public Tuple get(Key key) {
        try {
            IndexRecord indexRecord = index.get(key);

            if (indexRecord == null) {
                return null;
            }

            TupleBlock tupleBlock = getTupleBlock(indexRecord.blockOffset(), indexRecord.blockSize());
            Tuple read = tupleBlock.get(key);
            tupleBlock.memory().release();

            return read;
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public CloseableIterator<Tuple> ascendingIterator(long snapshotId) {
        return new LatestTupleIterator(snapshotId, new AscendingIterator(new AscendingBlockIterator(0)));
    }

    @Override
    public CloseableIterator<Tuple> descendingIterator(long snapshotId) {
        try {
            long startOffset = tableFile.size() - TableTrailer.SIZE - Sizes.INT_SIZE;
            return new LatestTupleIterator(snapshotId, new DescendingIterator(new DescendingBlockIterator
                    (startOffset)));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public CloseableIterator<Tuple> ascendingIterator(Key key, long snapshotId) {
        try {
            IndexRecord indexRecord = index.get(key);

            if (indexRecord == null) {
                return new CloseableIterator.Wrapper<Tuple>(Collections.<Tuple>emptyIterator());
            }

            TupleBlock startTupleBlock = readTupleBlock(indexRecord.blockOffset(), indexRecord.blockSize());
            Iterator<Tuple> startRecordIterator = startTupleBlock.ascendingIterator(key);
            long nextBlockOffset = indexRecord.blockOffset() + indexRecord.blockSize() + Sizes.INT_SIZE;
            return new LatestTupleIterator(snapshotId, new AscendingIterator(new AscendingBlockIterator
                    (nextBlockOffset), startRecordIterator, startTupleBlock));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public CloseableIterator<Tuple> descendingIterator(Key key, long snapshotId) {
        try {
            IndexRecord indexRecord = index.get(key);

            if (indexRecord == null) {
                return new CloseableIterator.Wrapper<Tuple>(Collections.<Tuple>emptyIterator());
            }

            TupleBlock startTupleBlock = readTupleBlock(indexRecord.blockOffset(), indexRecord.blockSize());
            Iterator<Tuple> startRecordIterator = startTupleBlock.descendingIterator(key);
            long nextBlockOffset = indexRecord.blockOffset() - Sizes.LONG_SIZE;
            return new LatestTupleIterator(snapshotId, new DescendingIterator(new DescendingBlockIterator
                    (nextBlockOffset), startRecordIterator, startTupleBlock));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public long id() {
        return tableId;
    }

    @Override
    public long tupleCount() {
        return trailer.recordCount();
    }

    @Override
    public long size() {
        return fileSize;
    }

    @Override
    public int level() {
        return trailer.level();
    }

    @Override
    public long maxSnapshotId() {
        return trailer.maxSnapshotId();
    }

    @Override
    public void close() {
        try {
            index.close();
            tableFile.close();
            tableBloomFilter.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public boolean isPersistent() {
        return true;
    }

    @Override
    public Iterator<Tuple> iterator() {
        return new AscendingIterator(new AscendingBlockIterator(0));
    }

    @Override
    public int compareTo(Table o) {
        return Long.compare(tableId, o.id());
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;

        FileTable fileTable = (FileTable) o;

        if (tableId != fileTable.tableId) return false;

        return true;
    }

    @Override
    public int hashCode() {
        return (int) (tableId ^ (tableId >>> 32));
    }

    @Override
    public String toString() {
        return "FileTable{" +
                "tableId=" + tableId +
                ", level=" + trailer.level() +
                '}';
    }

    private TupleBlock getTupleBlock(long offset, int size) throws IOException {
        TupleBlock tupleBlock = recordCache.get(tableId, offset);
        tableCacheHitRate.sample(tupleBlock != null);

        if (tupleBlock == null) {
            tupleBlock = readTupleBlock(offset, size);
            recordCache.put(tableId, offset, tupleBlock);
        }

        return tupleBlock;
    }

    private TupleBlock readTupleBlock(long offset, int size) throws IOException {
        MemoryPointer recordBlockPointer = MemoryAllocator.allocate(size);

        try {
            ByteBuffer recordBlockBuffer = recordBlockPointer.directBuffer();
            tableFile.read(recordBlockBuffer, offset);
            recordBlockBuffer.rewind();
            return new TupleBlock(new SortedByteMap(recordBlockPointer));
        } catch (IOException e) {
            recordBlockPointer.release();
            throw e;
        }
    }

    public static FileTable open(long tableId, Paths paths, TupleBlock.Cache recordCache,
                                 IndexBlock.Cache indexCache, Metrics metrics) throws IOException {
        Index index = Index.open(tableId, paths, indexCache, metrics);
        TableBloomFilter tableBloomFilter = TableBloomFilter.read(tableId, paths);
        ImmutableFile tableFile = ImmutableChannelFile.open(paths.tablePath(tableId));
        TableTrailer trailer = TableTrailer.read(tableFile);
        return new FileTable(tableId, index, tableBloomFilter, tableFile, trailer, recordCache, metrics);
    }
} public class FileTableWriter {

    /**
     * Provides a handy runnable for executing a self-contained FileTableWriter on a background thread.
     */
    public static class Task implements Runnable {

        public static class Builder {

            private long tableId;
            private int level = 1;
            private Iterator<Tuple> source;
            private long tupleCount;
            private Paths paths;
            private Config config;
            private Callback callback;
            private Throttle throttle = new Throttle(Integer.MAX_VALUE);

            public Builder tableId(long tableId) {
                this.tableId = tableId;
                return this;
            }

            public Builder level(int level) {
                this.level = level;
                return this;
            }

            public Builder source(Iterator<Tuple> source) {
                this.source = source;
                return this;
            }

            public Builder tupleCount(long tupleCount) {
                this.tupleCount = tupleCount;
                return this;
            }

            public Builder paths(Paths paths) {
                this.paths = paths;
                return this;
            }

            public Builder config(Config config) {
                this.config = config;
                return this;
            }

            public Builder callback(Callback callback) {
                this.callback = callback;
                return this;
            }

            public Builder throttle(Throttle throttle) {
                this.throttle = throttle;
                return this;
            }

            public Task build() {
                return new Task(tableId, level, paths, config, source, tupleCount, callback, throttle);
            }
        }

        public interface Callback {
            public void finish();
        }

        private static final AtomicInteger taskId = new AtomicInteger();
        private static final Logger logger = LoggerFactory.getLogger(FileTableWriter.class);

        private final long tableId;
        private final int level;
        private final Iterator<Tuple> tuples;
        private final long tupleCount;
        private final Paths paths;
        private final Config config;
        private final Callback callback;
        private final Throttle throttle;

        public Task(long tableId, int level, Paths paths, Config config, Iterator<Tuple> tuples, long tupleCount,
                    Callback callback, Throttle throttle) {
            this.tableId = tableId;
            this.level = level;
            this.paths = paths;
            this.config = config;
            this.tuples = tuples;
            this.tupleCount = tupleCount;
            this.callback = callback;
            this.throttle = throttle;
        }

        @Override
        public void run() {
            try {
                int id = taskId.incrementAndGet();
                logger.debug("Starting table writer " + id + " for table " + tableId);

                FileTableWriter tableWriter = FileTableWriter.open(tableId, paths, tupleCount,
                        config.indexBlockSize(), config.tableBlockSize(), level);

                while (tuples.hasNext()) {
                    Tuple tuple = tuples.next();
                    tableWriter.write(tuple);
                    throttle.consume(tuple.size());
                }

                tableWriter.finish();

                Files.move(paths.tempPath(tableId), paths.tablePath(tableId), StandardCopyOption.ATOMIC_MOVE);

                if (callback != null) {
                    callback.finish();
                }

                logger.debug("Finishing table writer " + id);
            } catch (ClosedChannelException e) {
                logger.debug("File table was only partially written " + tableId);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    private final int maxRecordBlockSize;
    private final IndexWriter indexWriter;
    private final TableBloomFilterWriter filterWriter;
    private final TableTrailer.Builder trailerBuilder;
    private final AppendFile tableDataFile;

    private TupleBlock.Builder recordBlockBuilder;

    private FileTableWriter(long tableId, IndexWriter indexWriter, TableBloomFilterWriter filterWriter,
                            AppendFile tableDataFile, int maxRecordBlockSize, int level) throws IOException {
        this.indexWriter = indexWriter;
        this.filterWriter = filterWriter;
        this.recordBlockBuilder = new TupleBlock.Builder();
        this.maxRecordBlockSize = maxRecordBlockSize;
        this.trailerBuilder = new TableTrailer.Builder(tableId, level);
        this.tableDataFile = tableDataFile;
    }

    public void write(Tuple tuple) throws IOException {
        if (recordBlockBuilder.size() >= maxRecordBlockSize) {
            writeRecordBlock();
        }

        recordBlockBuilder.addRecord(tuple);
        filterWriter.write(tuple.key());
        trailerBuilder.put(tuple);
    }

    public void finish() throws IOException {
        writeRecordBlock();
        writeTrailer();
        filterWriter.finish();
        indexWriter.finish();
        tableDataFile.close();
    }

    private void writeRecordBlock() throws IOException {
        TupleBlock tupleBlock = recordBlockBuilder.build();
        ByteBuffer recordBlockBuffer = tupleBlock.memory().directBuffer();

        tableDataFile.appendInt(recordBlockBuffer.capacity());
        long recordBlockOffset = tableDataFile.append(recordBlockBuffer);
        recordBlockBuffer.rewind();
        tableDataFile.appendInt(recordBlockBuffer.capacity());

        Tuple startTuple = tupleBlock.first();
        indexWriter.write(new IndexRecord(startTuple.key(), recordBlockOffset, recordBlockBuffer.capacity()));
        tupleBlock.memory().release();
        recordBlockBuilder = new TupleBlock.Builder();
    }

    private void writeTrailer() throws IOException {
        ByteBuffer trailerBuffer = trailerBuilder.build().buffer();
        tableDataFile.append(trailerBuffer);
    }

    public static FileTableWriter open(long tableId, Paths paths, long approxRecordCount, int maxIndexBlockSize,
                                       int maxRecordBlockSize, int level) throws IOException {
        IndexWriter indexWriter = IndexWriter.open(tableId, paths, maxIndexBlockSize);
        TableBloomFilterWriter filterWriter = TableBloomFilterWriter.open(tableId, paths, approxRecordCount);
        AppendFile tableDataFile = AppendChannelFile.open(paths.tempPath(tableId));

        return new FileTableWriter(tableId, indexWriter, filterWriter, tableDataFile, maxRecordBlockSize, level);
    }
} public class TupleBlock implements Iterable<Tuple>, Offheap {

    public static class Cache {

        private final TableBlockCache<TupleBlock> cache;

        public Cache(long maxSize, Metrics metrics) {
            cache = new TableBlockCache<TupleBlock>(maxSize, new Weigher<TupleBlock>() {
                @Override
                public int weightOf(TupleBlock tuple) {
                    return tuple.memory().size();
                }
            });

            metrics.gauge("cache.tupleBlock.entrySize", new Gauge<Long>() {
                @Override
                public Long getValue() {
                    return cache.totalEntrySize();
                }
            });

            metrics.gauge("cache.tupleBlock.utilizationRate", new Gauge<Double>() {
                @Override
                public Double getValue() {
                    return cache.utilizationRate();
                }
            });
        }

        public TupleBlock get(long tableId, long offset) {
            return cache.get(tableId, offset);
        }

        public void put(long tableId, long offset, TupleBlock tupleBlock) {
            cache.put(tableId, offset, tupleBlock);
        }

        public void invalidate(long tableId) {
            cache.invalidate(tableId);
        }

        public void clear() {
            cache.clear();
        }
    }

    public static class Builder {

        private final SortedByteMap.Builder byteMapBuilder = new SortedByteMap.Builder();
        private int size;

        public void addRecord(Tuple tuple) {
            byteMapBuilder.add(new Key(tuple.key().data(), tuple.key().snapshotId()), tuple.value());
            size += tuple.size();
        }

        public int size() {
            return size;
        }

        public TupleBlock build() {
            return new TupleBlock(byteMapBuilder.build());
        }
    }

    private class TupleIterator implements Iterator<Tuple> {

        private final Iterator<SortedByteMap.Entry> entryIterator;

        private TupleIterator(Iterator<SortedByteMap.Entry> entryIterator) {
            this.entryIterator = entryIterator;
        }

        @Override
        public boolean hasNext() {
            return entryIterator.hasNext();
        }

        @Override
        public Tuple next() {
            SortedByteMap.Entry nextEntry = entryIterator.next();
            return new Tuple(nextEntry.key(), nextEntry.value());
        }

        @Override
        public void remove() {
            entryIterator.remove();
        }
    }

    private final SortedByteMap sortedByteMap;

    public TupleBlock(SortedByteMap sortedByteMap) {
        this.sortedByteMap = sortedByteMap;
    }

    public Tuple get(Key key) {
        int closestIndex = sortedByteMap.floorIndex(key);

        if (closestIndex < 0 || closestIndex >= sortedByteMap.entryCount()) {
            return null;
        }

        Tuple closestTuple = deserialize(closestIndex);
        return closestTuple.key().data().equals(key.data()) ? closestTuple : null;
    }

    public Tuple first() {
        return deserialize(0);
    }

    public Iterator<Tuple> ascendingIterator() {
        return new TupleIterator(sortedByteMap.ascendingIterator());
    }

    public Iterator<Tuple> ascendingIterator(Key key) {
        return new TupleIterator(sortedByteMap.ascendingIterator(key));
    }

    public Iterator<Tuple> descendingIterator() {
        return new TupleIterator(sortedByteMap.descendingIterator());
    }

    public Iterator<Tuple> descendingIterator(Key key) {
        return new TupleIterator(sortedByteMap.descendingIterator(key));
    }

    @Override
    public Iterator<Tuple> iterator() {
        return new TupleIterator(sortedByteMap.ascendingIterator());
    }

    @Override
    public MemoryPointer memory() {
        return sortedByteMap.memory();
    }

    @Override
    public String toString() {
        List<Tuple> tuples = new ArrayList<Tuple>();
        for (Tuple tuple : this) {
            tuples.add(tuple);
        }

        return "TupleBlock{tuples=" + tuples + "}";
    }

    private Tuple deserialize(int index) {
        SortedByteMap.Entry entry = sortedByteMap.get(index);
        return new Tuple(entry.key(), entry.value());
    }
} public class MemoryTable implements MutableTable {

    private final long id;
    private final SortedTupleMap records = new SkipListTupleMap();
    private final AtomicLong maxSnapshotId = new AtomicLong();
    private final AtomicInteger recordCount = new AtomicInteger();
    private final AtomicInteger size = new AtomicInteger();

    public MemoryTable(long id) {
        this.id = id;
    }

    @Override
    public void put(Tuple tuple) {
        records.put(tuple.key(), tuple.value());
        recordCount.incrementAndGet();
        size.addAndGet(tuple.size());
        maxSnapshotId.set(tuple.key().snapshotId());
    }

    @Override
    public long id() {
        return id;
    }

    @Override
    public boolean mightContain(Key key) {
        return get(key) != null;
    }

    @Override
    public Tuple get(Key key) {
        return records.get(key);
    }

    @Override
    public CloseableIterator<Tuple> ascendingIterator(long snapshotId) {
        return records.ascendingIterator(snapshotId);
    }

    @Override
    public CloseableIterator<Tuple> descendingIterator(long snapshotId) {
        return records.descendingIterator(snapshotId);
    }

    @Override
    public CloseableIterator<Tuple> ascendingIterator(Key key, long snapshotId) {
        return records.ascendingIterator(key, snapshotId);
    }

    @Override
    public CloseableIterator<Tuple> descendingIterator(Key key, long snapshotId) {
        return records.descendingIterator(key, snapshotId);
    }

    @Override
    public long tupleCount() {
        return recordCount.get();
    }

    @Override
    public long size() {
        return size.get();
    }

    @Override
    public int level() {
        return 0;
    }

    @Override
    public long maxSnapshotId() {
        return maxSnapshotId.get();
    }

    @Override
    public void close() {
    }

    @Override
    public boolean isPersistent() {
        return false;
    }

    @Override
    public Iterator<Tuple> iterator() {
        return records.iterator();
    }

    @Override
    public int compareTo(Table o) {
        return Long.compare(id, o.id());
    }
} public class CommitLog implements Iterable<Tuple>, Closeable {

    private class LogIterator implements Iterator<Tuple> {

        private final XORShiftRandom pseudoRandom;
        private final Queue<Tuple> nextTuple = new LinkedList<Tuple>();
        private long fileOffset = Sizes.LONG_SIZE;

        public LogIterator() {
            this.pseudoRandom = new XORShiftRandom(CommitLog.this.seed);
        }

        @Override
        public boolean hasNext() {
            if (!nextTuple.isEmpty()) {
                return true;
            }

            Tuple next = nextRecord();

            if (next == null) {
                return false;
            }

            nextTuple.add(next);

            return true;
        }

        @Override
        public Tuple next() {
            if (nextTuple.isEmpty()) {
                if (!hasNext()) {
                    throw new NoSuchElementException();
                }
            }

            return nextTuple.poll();
        }

        @Override
        public void remove() {
            throw new UnsupportedOperationException();
        }

        private Tuple nextRecord() {
            try {
                if (fileOffset >= logFile.size()) {
                    return null;
                }

                int recordSize = logFile.readInt(fileOffset);
                fileOffset += Sizes.INT_SIZE;

                ByteBuffer recordBuffer = ByteBuffer.allocate(recordSize);
                logFile.read(recordBuffer, fileOffset);
                fileOffset += recordSize;
                int nextInt = logFile.readInt(fileOffset);
                fileOffset += Sizes.INT_SIZE;

                if (nextInt != pseudoRandom.nextInt()) {
                    return null;
                }

                recordBuffer.rewind();

                return Tuple.SERIALIZER.deserialize(recordBuffer);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    private final long tableId;
    private final ImmutableFile logFile;
    private final long seed;

    private CommitLog(long tableId, ImmutableFile logFile) throws IOException {
        this.tableId = tableId;
        this.logFile = logFile;
        this.seed = getSeed();
    }

    public long tableId() {
        return tableId;
    }

    @Override
    public void close() throws IOException {
        logFile.close();
    }

    @Override
    public Iterator<Tuple> iterator() {
        return new LogIterator();
    }

    private long getSeed() throws IOException {
        return logFile.readLong(0);
    }

    public static CommitLog open(long tableId, Paths paths) throws IOException {
        ImmutableFile logFile = ImmutableChannelFile.open(paths.logPath(tableId));
        return new CommitLog(tableId, logFile);
    }
}
import java.nio.file.Files;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import com.jordanwilliams.heftydb.data.Tuple;
import com.jordanwilliams.heftydb.index.IndexBlock;
import com.jordanwilliams.heftydb.metrics.Metrics;
import com.jordanwilliams.heftydb.state.Caches;
import com.jordanwilliams.heftydb.state.Paths;
import com.jordanwilliams.heftydb.table.MutableTable;
import com.jordanwilliams.heftydb.table.Table;
import com.jordanwilliams.heftydb.table.file.FileTable;
import com.jordanwilliams.heftydb.table.file.FileTableWriter;
import com.jordanwilliams.heftydb.table.file.TupleBlock;
import com.jordanwilliams.heftydb.table.memory.MemoryTable;
import com.jordanwilliams.heftydb.write.CommitLog;
import java.io.IOException;
/*
 * Copyright (c) 2014. Jordan Williams
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.jordanwilliams.heftydb.db;



/**
 * Restores the state of an existing database and does any clean up needed to get into a consistent state.
 */
public class DBInitializer {

    private final Config config;
    private final Paths paths;
    private final Caches caches;
    private final Metrics metrics;
    private long maxSnapshotId;

    public DBInitializer(Config config, Metrics metrics) {
        this.config = config;
        this.metrics = metrics;
        this.paths = new Paths(config.tableDirectory(), config.logDirectory());
        this.caches = new Caches(new TupleBlock.Cache(config.tableCacheSize(), metrics),
                new IndexBlock.Cache(config.indexCacheSize(), metrics));
    }

    public DBState initialize() throws IOException {
        deleteTempTables();
        writeTablesFromLogs();
        List<Table> tables = loadTables();
        return new DBState(tables, config, paths, caches, maxSnapshotId);
    }

    private List<Table> loadTables() throws IOException {
        List<Table> tables = new ArrayList<Table>();
        Set<Long> tableIds = paths.tableFileIds();

        for (Long id : tableIds) {
            Table table = FileTable.open(id, paths, caches.recordBlockCache(), caches.indexBlockCache(), metrics);
            maxSnapshotId = Math.max(table.maxSnapshotId(), maxSnapshotId);
            tables.add(table);
        }

        return tables;
    }

    private void deleteTempTables() throws IOException {
        Set<Long> tempIds = paths.tempTableFileIds();

        for (Long id : tempIds) {
            Files.deleteIfExists(paths.tempPath(id));
            Files.deleteIfExists(paths.indexPath(id));
            Files.deleteIfExists(paths.filterPath(id));
        }
    }

    private void writeTablesFromLogs() throws IOException {
        Set<Long> logIds = paths.logFileIds();

        for (Long id : logIds) {
            CommitLog log = CommitLog.open(id, paths);
            Table memoryTable = readTable(log);
            log.close();

            FileTableWriter.Task tableWriterTask = new FileTableWriter.Task.Builder().tableId(id).config(config)