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 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 Index {

    private static final int ROOT_INDEX_BLOCK_OFFSET = 8;
    private static final int ROOT_INDEX_BLOCK_SIZE_OFFSET = 12;

    private final long tableId;
    private final ImmutableFile indexFile;
    private final IndexBlock.Cache cache;
    private final Metrics metrics;
    private final IndexBlock rootIndexBlock;

    private final Histogram indexSearchLevels;
    private final CacheHitGauge indexCacheHitRate;

    private Index(long tableId, ImmutableFile indexFile, IndexBlock.Cache cache, Metrics metrics) throws IOException {
        this.tableId = tableId;
        this.indexFile = indexFile;
        this.cache = cache;
        this.metrics = metrics;
        long rootBlockOffset = indexFile.readLong(indexFile.size() - ROOT_INDEX_BLOCK_OFFSET);
        int rootBlockSize = indexFile.readInt(indexFile.size() - ROOT_INDEX_BLOCK_SIZE_OFFSET);
        this.rootIndexBlock = readIndexBlock(rootBlockOffset, rootBlockSize);

        this.indexSearchLevels = metrics.histogram("index.searchLevels");
        this.indexCacheHitRate = metrics.hitGauge("index.cacheHitRate");
    }

    public IndexRecord get(Key key) throws IOException {
        IndexRecord currentIndexRecord = rootIndexBlock.get(key);
        int searchLevels = 1;

        while (currentIndexRecord != null && !currentIndexRecord.isLeaf()) {
            IndexBlock currentIndexBlock = getIndexBlock(currentIndexRecord.blockOffset(),
                    currentIndexRecord.blockSize());
            currentIndexRecord = currentIndexBlock.get(key);
            currentIndexBlock.memory().release();
            searchLevels++;
        }

        indexSearchLevels.update(searchLevels);

        return currentIndexRecord;
    }

    public void close() throws IOException {
        rootIndexBlock.memory().release();
        indexFile.close();
        cache.clear();
    }

    private IndexBlock getIndexBlock(long blockOffset, int blockSize) throws IOException {
        IndexBlock indexBlock = cache.get(tableId, blockOffset);
        indexCacheHitRate.sample(indexBlock != null);

        if (indexBlock == null) {
            indexBlock = readIndexBlock(blockOffset, blockSize);
            cache.put(tableId, blockOffset, indexBlock);
        }

        return indexBlock;
    }

    private IndexBlock readIndexBlock(long blockOffset, int blockSize) throws IOException {
        MemoryPointer indexPointer = MemoryAllocator.allocate(blockSize);

        try {
            ByteBuffer indexBuffer = indexPointer.directBuffer();
            indexFile.read(indexBuffer, blockOffset);
            indexBuffer.rewind();
            return new IndexBlock(new SortedByteMap(indexPointer));
        } catch (IOException e) {
            indexPointer.release();
            throw e;
        }
    }

    public static Index open(long tableId, Paths paths, IndexBlock.Cache cache, Metrics metrics) throws IOException {
        ImmutableFile indexFile = ImmutableChannelFile.open(paths.indexPath(tableId));
        return new Index(tableId, indexFile, cache, metrics);
    }
} 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 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 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 TupleGenerator {

    public interface Function<T> {
        public T apply();
    }

    private final KeyValueGenerator testDataGenerator = new KeyValueGenerator();

    public List<Tuple> testRecords(int startingSnapshotId, int recordCount, int keyReuse, Function<Integer> keySize,
                                   Function<Integer> valueSize) {
        int snapshotId = startingSnapshotId;
        List<Tuple> tuples = new ArrayList<Tuple>(recordCount);

        for (int i = 0; i < recordCount; i++) {
            ByteBuffer key = testDataGenerator.testKey(keySize.apply(), keyReuse);
            ByteBuffer value = testDataGenerator.testValue(valueSize.apply());
            tuples.add(new Tuple(new Key(key, snapshotId), new Value(value)));
            snapshotId++;
        }

        Collections.sort(tuples);

        return tuples;
    }


    public List<Tuple> testRecords(int startingSnapshotId, int recordCount, int keyReuse, final int keySize,
                                   final int valueSize) {
        return testRecords(startingSnapshotId, recordCount, keyReuse, new Function<Integer>() {
                    @Override
                    public Integer apply() {
                        return keySize;
                    }
                }, new Function<Integer>() {
                    @Override
                    public Integer apply() {
                        return valueSize;
                    }
                }
        );
    }

    public List<Tuple> testRecords(int recordCount, int keyReuse, int keySize, int valueSize) {
        return testRecords(0, recordCount, keyReuse, keySize, valueSize);
    }

    public List<Tuple> testRecords(int recordCount, int keyReuse) {
        return testRecords(recordCount, keyReuse, 16, 100);
    }

    public static List<Tuple> latest(List<Tuple> tuples, long snapshotId) {
        SortedMap<ByteBuffer, Tuple> latestRecordMap = new TreeMap<ByteBuffer, Tuple>();

        for (Tuple tuple : tuples) {
            Tuple existing = latestRecordMap.get(tuple.key().data());

            if (existing == null || tuple.key().snapshotId() > existing.key().snapshotId()) {
                latestRecordMap.put(tuple.key().data(), tuple);
            }
        }

        return new ArrayList<Tuple>(latestRecordMap.values());
    }
} 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);
        }
    }
}
import com.jordanwilliams.heftydb.data.Tuple;
import com.jordanwilliams.heftydb.db.Config;
import com.jordanwilliams.heftydb.index.Index;
import com.jordanwilliams.heftydb.index.IndexBlock;
import com.jordanwilliams.heftydb.metrics.Metrics;
import com.jordanwilliams.heftydb.state.Paths;
import com.jordanwilliams.heftydb.table.file.FileTableWriter;
import com.jordanwilliams.heftydb.test.generator.ConfigGenerator;
import com.jordanwilliams.heftydb.test.generator.TupleGenerator;
import com.jordanwilliams.heftydb.test.helper.TestFileHelper;
import java.util.List;
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 IndexPerformance {

    public static void main(String[] args) throws Exception {
        TestFileHelper.createTestDirectory();
        TupleGenerator generator = new TupleGenerator();
        List<Tuple> tuples = generator.testRecords(1, 500000, 20, 16, 100);

        Paths paths = ConfigGenerator.testPaths();
        Config config = ConfigGenerator.testConfig();