public class Key implements Comparable<Key> {

    private final ByteBuffer data;
    private final long snapshotId;

    public Key(ByteBuffer data, long snapshotId) {
        this.data = data;
        this.snapshotId = snapshotId;
    }

    public ByteBuffer data() {
        return data;
    }

    public long snapshotId() {
        return snapshotId;
    }

    public int size() {
        return data.capacity();
    }

    @Override
    public int compareTo(Key o) {
        int compared = data.compareTo(o.data);

        if (compared != 0) {
            return compared;
        }

        return Long.compare(snapshotId, o.snapshotId);
    }

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

        Key key = (Key) o;

        if (snapshotId != key.snapshotId) return false;
        if (data != null ? !data.equals(key.data) : key.data != null) return false;

        return true;
    }

    @Override
    public int hashCode() {
        return data != null ? data.hashCode() : 0;
    }

    @Override
    public String toString() {
        byte[] keyArray = new byte[data.capacity()];
        data.rewind();
        data.get(keyArray);
        data.rewind();

        return "Key{" +
                "data=" + new String(keyArray) +
                ", snapshotId=" + snapshotId +
                '}';
    }
} 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 Value implements Comparable<Value> {

    public static Value TOMBSTONE_VALUE = new Value(ByteBuffers.EMPTY_BUFFER);

    private final ByteBuffer value;

    public Value(ByteBuffer value) {
        this.value = value;
    }

    public ByteBuffer data() {
        return value;
    }

    public boolean isEmpty() {
        return value.capacity() == 0;
    }

    public int size() {
        return value.capacity();
    }

    @Override
    public int compareTo(Value o) {
        return value.compareTo(o.value);
    }

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

        Value value1 = (Value) o;

        if (value != null ? !value.equals(value1.value) : value1.value != null) {
            return false;
        }

        return true;
    }

    @Override
    public int hashCode() {
        return value != null ? value.hashCode() : 0;
    }

    @Override
    public String toString() {
        return "Value{" +
                "data=" + new String(value.array()) +
                "}";
    }
} public class Config {

    public static class Builder {

        //Default config values
        private CompactionStrategy compactionStrategy = CompactionStrategies.SIZE_TIERED_COMPACTION_STRATEGY;
        private int memoryTableSize = 8192000;
        private int tableBlockSize = 16384;
        private int indexBlockSize = 65536;
        private int tableWriterThreads = 4;
        private int tableCompactionThreads = 8;
        private long tableCacheSize = 128000000;
        private long indexCacheSize = 32000000;
        private long maxCompactionRate = 32768000;
        private long maxWriteRate = 32768000;
        private boolean printMetrics = false;
        private Path tableDirectory;
        private Path logDirectory;

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

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

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

        public Builder compactionStrategy(CompactionStrategy compactionStrategy) {
            this.compactionStrategy = compactionStrategy;
            return this;
        }

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

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

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

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

        public Builder printMetrics(boolean printMetrics) {
            this.printMetrics = printMetrics;
            return this;
        }

        public Builder directory(Path directory) {
            this.tableDirectory = directory;
            this.logDirectory = directory;
            return this;
        }

        public Builder tableDirectory(Path tableDirectory) {
            this.tableDirectory = tableDirectory;
            return this;
        }

        public Builder logDirectory(Path logDirectory) {
            this.logDirectory = logDirectory;
            return this;
        }

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

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

        public Config build() {
            return new Config(compactionStrategy, memoryTableSize, tableBlockSize, indexBlockSize,
                    tableWriterThreads, tableCompactionThreads, tableCacheSize, indexCacheSize, printMetrics,
                    tableDirectory, logDirectory, maxCompactionRate, maxWriteRate);
        }
    }

    private final CompactionStrategy compactionStrategy;
    private final int memoryTableSize;
    private final int tableBlockSize;
    private final int indexBlockSize;
    private final int tableWriterThreads;
    private final int tableCompactionThreads;
    private final long tableCacheSize;
    private final long indexCacheSize;
    private final boolean printMetrics;
    private final Path tableDirectory;
    private final Path logDirectory;
    private final long maxCompactionRate;
    private final long maxWriteRate;

    public Config(CompactionStrategy compactionStrategy, int memoryTableSize, int tableBlockSize, int indexBlockSize,
                  int tableWriterThreads, int tableCompactionThreads, long tableCacheSize, long indexCacheSize,
                  boolean printMetrics, Path tableDirectory, Path logDirectory, long maxCompactionRate,
                  long maxWriteRate) {
        this.compactionStrategy = compactionStrategy;
        this.memoryTableSize = memoryTableSize;
        this.tableBlockSize = tableBlockSize;
        this.indexBlockSize = indexBlockSize;
        this.tableWriterThreads = tableWriterThreads;
        this.tableCompactionThreads = tableCompactionThreads;
        this.tableCacheSize = tableCacheSize;
        this.indexCacheSize = indexCacheSize;
        this.printMetrics = printMetrics;
        this.tableDirectory = tableDirectory;
        this.logDirectory = logDirectory;
        this.maxCompactionRate = maxCompactionRate;
        this.maxWriteRate = maxWriteRate;
    }

    public CompactionStrategy compactionStrategy() {
        return compactionStrategy;
    }

    public int memoryTableSize() {
        return memoryTableSize;
    }

    public int tableBlockSize() {
        return tableBlockSize;
    }

    public int indexBlockSize() {
        return indexBlockSize;
    }

    public int tableWriterThreads() {
        return tableWriterThreads;
    }

    public int tableCompactionThreads() {
        return tableCompactionThreads;
    }

    public long tableCacheSize() {
        return tableCacheSize;
    }

    public long indexCacheSize() {
        return indexCacheSize;
    }

    public boolean autoPrintMetrics() {
        return printMetrics;
    }

    public Path tableDirectory() {
        return tableDirectory;
    }

    public Path logDirectory() {
        return logDirectory;
    }

    public long maxCompactionRate() {
        return maxCompactionRate;
    }

    public long maxWriteRate() {
        return maxWriteRate;
    }

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

        Config config = (Config) o;

        if (indexBlockSize != config.indexBlockSize) return false;
        if (indexCacheSize != config.indexCacheSize) return false;
        if (maxCompactionRate != config.maxCompactionRate) return false;
        if (maxWriteRate != config.maxWriteRate) return false;
        if (memoryTableSize != config.memoryTableSize) return false;
        if (printMetrics != config.printMetrics) return false;
        if (tableBlockSize != config.tableBlockSize) return false;
        if (tableCacheSize != config.tableCacheSize) return false;
        if (tableCompactionThreads != config.tableCompactionThreads) return false;
        if (tableWriterThreads != config.tableWriterThreads) return false;
        if (compactionStrategy != null ? !compactionStrategy.equals(config.compactionStrategy) : config
                .compactionStrategy != null)
            return false;
        if (logDirectory != null ? !logDirectory.equals(config.logDirectory) : config.logDirectory != null)
            return false;
        if (tableDirectory != null ? !tableDirectory.equals(config.tableDirectory) : config.tableDirectory != null)
            return false;

        return true;
    }

    @Override
    public int hashCode() {
        int result = compactionStrategy != null ? compactionStrategy.hashCode() : 0;
        result = 31 * result + memoryTableSize;
        result = 31 * result + tableBlockSize;
        result = 31 * result + indexBlockSize;
        result = 31 * result + tableWriterThreads;
        result = 31 * result + tableCompactionThreads;
        result = 31 * result + (int) (tableCacheSize ^ (tableCacheSize >>> 32));
        result = 31 * result + (int) (indexCacheSize ^ (indexCacheSize >>> 32));
        result = 31 * result + (printMetrics ? 1 : 0);
        result = 31 * result + (tableDirectory != null ? tableDirectory.hashCode() : 0);
        result = 31 * result + (logDirectory != null ? logDirectory.hashCode() : 0);
        result = 31 * result + (int) (maxCompactionRate ^ (maxCompactionRate >>> 32));
        result = 31 * result + (int) (maxWriteRate ^ (maxWriteRate >>> 32));
        return result;
    }

    @Override
    public String toString() {
        return "Config{" +
                "compactionStrategy=" + compactionStrategy +
                ", memoryTableSize=" + memoryTableSize +
                ", tableBlockSize=" + tableBlockSize +
                ", indexBlockSize=" + indexBlockSize +
                ", tableWriterThreads=" + tableWriterThreads +
                ", tableCompactionThreads=" + tableCompactionThreads +
                ", tableCacheSize=" + tableCacheSize +
                ", indexCacheSize=" + indexCacheSize +
                ", autoPrintMetrics=" + printMetrics +
                ", tableDirectory=" + tableDirectory +
                ", logDirectory=" + logDirectory +
                ", maxCompactionRate=" + maxCompactionRate +
                ", maxWriteRate=" + maxWriteRate +
                '}';
    }
} 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 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 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 ConfigGenerator {

    public static Paths testPaths() {
        return new Paths(TestFileHelper.TEMP_PATH, TestFileHelper.TEMP_PATH);
    }

    public static Caches testCaches() {
        return new Caches(new TupleBlock.Cache(32768000, new Metrics(testConfig())), new IndexBlock.Cache(16384000,
                new Metrics(testConfig())));
    }

    public static Config defaultConfig() {
        return new Config.Builder().tableDirectory(TestFileHelper.TEMP_PATH).build();
    }

    public static Config enduranceConfig() {
        return new Config.Builder().tableDirectory(TestFileHelper.TEMP_PATH).printMetrics(true).indexCacheSize
                (128000000).tableCacheSize(512000000).memoryTableSize(8192000).maxWriteRate(32768000).build();
    }

    public static Config performanceConfig() {
        return new Config.Builder().tableDirectory(TestFileHelper.TEMP_PATH).indexCacheSize(128000000).tableCacheSize
                (512000000).tableBlockSize(1024).memoryTableSize(8192000).maxCompactionRate(32768000).build();
    }

    public static Config testConfig() {
        Config.Builder builder = new Config.Builder();

        return builder.compactionStrategy(CompactionStrategies.NULL_COMPACTION_STRATEGY).memoryTableSize(16384)
                .tableBlockSize(4096).indexBlockSize(4096).tableCacheSize(1024000).indexCacheSize(1024000)
                .tableDirectory(TestFileHelper.TEMP_PATH).build();
    }

    public static DBState perfState() {
        return new DBState(Collections.<Table>emptyList(), defaultConfig(), testPaths(), testCaches(), 1);
    }
} public class KeyValueGenerator {

    private static final int MAX_CACHED_KEYS = 1024;

    private final Random rand = new Random(System.nanoTime());
    private final List<ByteBuffer> testKeys = new ArrayList<ByteBuffer>();
    private final Map<Integer, ByteBuffer> testValues = new HashMap<Integer, ByteBuffer>();

    public synchronized ByteBuffer testKey(int size, int reuseWeight) {
        int next = rand.nextInt(100);

        if (next <= reuseWeight && reuseWeight != 0 && testKeys.size() > 0) {
            ByteBuffer existing = testKeys.get(rand.nextInt(testKeys.size()));
            ByteBuffer reusedKey = existing.duplicate();
            reusedKey.rewind();
            return reusedKey;
        } else {
            ByteBuffer random = randomKey(size);

            if (testKeys.size() < MAX_CACHED_KEYS) {
                testKeys.add(random);
            }

            return random.duplicate();
        }
    }

    public ByteBuffer testKey(int reuseWeight) {
        return testKey(16, reuseWeight);
    }

    public synchronized ByteBuffer testValue(int size) {
        if (!testValues.containsKey(size)) {
            testValues.put(size, randomValue(size));
        }

        ByteBuffer testValue = testValues.get(size).duplicate();
        testValue.rewind();
        return testValue;
    }

    private ByteBuffer randomKey(int size) {
        return ByteBuffer.wrap(RandomStringUtils.randomAlphanumeric(size).getBytes());
    }

    private ByteBuffer randomValue(int size) {
        String randomString = RandomStringUtils.randomAlphanumeric(size);
        return ByteBuffer.wrap(randomString.getBytes(Charset.defaultCharset()));
    }
} public class PerformanceHelper {

    public static ConsoleReporter consoleReporter(MetricRegistry metrics) {
        return ConsoleReporter.forRegistry(metrics).convertDurationsTo(TimeUnit.MICROSECONDS).convertRatesTo(TimeUnit
                .SECONDS).build();
    }

} public class TestFileHelper {

    public static final Path TEMP_PATH = Paths.get("/tmp/heftytest");

    public static void cleanUpTestFiles() throws IOException {
        DirectoryStream<Path> filePaths = Files.newDirectoryStream(TEMP_PATH);

        for (Path f : filePaths) {
            if (Files.exists(f)) {
                Files.delete(f);
            }
        }
    }

    public static void createTestDirectory() throws IOException {
        if (!Files.exists(TEMP_PATH)) {
            Files.createDirectory(TEMP_PATH);
        }
    }
} public class ByteBuffers {

    public static final ByteBuffer EMPTY_BUFFER = ByteBuffer.allocate(0);

    public static ByteBuffer fromString(String string) {
        return ByteBuffer.wrap(string.getBytes(Charset.defaultCharset()));
    }

    public static String toString(ByteBuffer byteBuffer) {
        byte[] bytes = new byte[byteBuffer.capacity()];

        for (int i = 0; i < byteBuffer.capacity(); i++) {
            bytes[i] = byteBuffer.get(i);
        }

        return new String(bytes);
    }
}
import com.codahale.metrics.ConsoleReporter;
import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.Timer;
import com.jordanwilliams.heftydb.data.Key;
import com.jordanwilliams.heftydb.data.Tuple;
import com.jordanwilliams.heftydb.data.Value;
import com.jordanwilliams.heftydb.db.Config;
import com.jordanwilliams.heftydb.index.IndexBlock;
import com.jordanwilliams.heftydb.metrics.Metrics;
import com.jordanwilliams.heftydb.state.Paths;
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.test.generator.ConfigGenerator;
import com.jordanwilliams.heftydb.test.generator.KeyValueGenerator;
import com.jordanwilliams.heftydb.test.helper.PerformanceHelper;
import com.jordanwilliams.heftydb.test.helper.TestFileHelper;
import com.jordanwilliams.heftydb.util.ByteBuffers;
import java.nio.file.Files;
import java.nio.file.StandardCopyOption;
import java.util.Random;
/*
 * 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.test.performance.table.file;



public class FileTablePerformance {

    private static final int RECORD_COUNT = 5 * 1000000;

    public static void main(String[] args) throws Exception {
        MetricRegistry metrics = new MetricRegistry();
        ConsoleReporter reporter = PerformanceHelper.consoleReporter(metrics);
        Timer writeTimer = metrics.timer("writes");
        Timer readTimer = metrics.timer("reads");
        Config config = ConfigGenerator.performanceConfig();