public class Compactor {

    private static final Logger logger = LoggerFactory.getLogger(Compactor.class);

    private class Task implements Runnable {

        private final CompactionTask compactionTask;
        private final Throttle throttle;

        private Task(CompactionTask compactionTask, Throttle throttle) {
            this.compactionTask = compactionTask;
            this.throttle = throttle;
        }

        @Override
        public void run() {
            try {
                Timer.Context watch = metrics.timer("compactor.taskExecution").time();
                List<CloseableIterator<Tuple>> tableIterators = new ArrayList<CloseableIterator<Tuple>>();
                long tupleCount = 0;
                long nextTableId = tables.nextId();

                for (Table table : compactionTask.tables()) {
                    tableIterators.add(new CloseableIterator.Wrapper<Tuple>(table.iterator()));
                    tupleCount += table.tupleCount();
                }

                Iterator<Tuple> compactionIterator = new CompactionTupleIterator(snapshots.minimumRetainedId(),
                        new MergingIterator<Tuple>
                        (tableIterators));

                FileTableWriter.Task writerTask = new FileTableWriter.Task.Builder().tableId(nextTableId).config
                        (config).paths(paths).level(compactionTask.level()).tupleCount(tupleCount).source
                        (compactionIterator).throttle(throttle).build();

                writerTask.run();

                tables.add(FileTable.open(nextTableId, paths, caches.recordBlockCache(), caches.indexBlockCache(),
                        metrics));

                removeObsoleteTables(compactionTask.tables());

                watch.stop();
            } catch (ClosedChannelException e) {
                logger.debug("Compaction terminated without finishing " + compactionId);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }

        private void removeObsoleteTables(List<Table> toRemove) throws IOException {
            tables.removeAll(toRemove);

            for (Table table : toRemove) {
                table.close();
                caches.indexBlockCache().invalidate(table.id());
                caches.recordBlockCache().invalidate(table.id());
                Files.deleteIfExists(paths.tablePath(table.id()));
                Files.deleteIfExists(paths.indexPath(table.id()));
                Files.deleteIfExists(paths.filterPath(table.id()));
            }
        }
    }

    private final Config config;
    private final Paths paths;
    private final Tables tables;
    private final CompactionTables compactionTables;
    private final Caches caches;
    private final ThreadPoolExecutor compactionExecutor;
    private final ThreadPoolExecutor compactionTaskExecutor;
    private final ThreadPoolExecutor highPriorityCompactionTaskExecutor;
    private final CompactionPlanner compactionPlanner;
    private final Metrics metrics;
    private final AtomicInteger compactionId = new AtomicInteger();
    private final Snapshots snapshots;

    public Compactor(Config config, Paths paths, Tables tables, Caches caches, CompactionStrategy compactionStrategy,
                     Metrics metrics, Snapshots snapshots) {
        this.config = config;
        this.paths = paths;
        this.tables = tables;
        this.compactionTables = new CompactionTables(tables);
        this.caches = caches;
        this.metrics = metrics;
        this.snapshots = snapshots;

        this.compactionExecutor = new ThreadPoolExecutor(config.tableCompactionThreads(), config.tableCompactionThreads(),
                Long.MAX_VALUE, TimeUnit.DAYS, new LinkedBlockingQueue<Runnable>(config.tableCompactionThreads()),
                new ThreadFactoryBuilder().setNameFormat("Compaction thread %d").build(), new ThreadPoolExecutor.CallerRunsPolicy());

        int compactionTaskThreads = Math.max(config.tableCompactionThreads() / 2, 1);

        this.compactionTaskExecutor = new ThreadPoolExecutor(compactionTaskThreads, compactionTaskThreads, Long.MAX_VALUE,
                TimeUnit.DAYS, new LinkedBlockingQueue<Runnable>(config.tableCompactionThreads()),
                new ThreadFactoryBuilder().setNameFormat("Compaction task thread %d").build(),
                new ThreadPoolExecutor.CallerRunsPolicy());

        this.highPriorityCompactionTaskExecutor = new ThreadPoolExecutor(compactionTaskThreads, compactionTaskThreads,
                Long.MAX_VALUE, TimeUnit.DAYS, new LinkedBlockingQueue<Runnable>(config.tableCompactionThreads()),
                new ThreadFactoryBuilder().setNameFormat("High priority " +
                "compaction task thread %d").build(), new ThreadPoolExecutor.CallerRunsPolicy());

        this.compactionPlanner = compactionStrategy.initialize(compactionTables);

        tables.addChangeHandler(new Tables.ChangeHandler() {
            @Override
            public void changed() {
                evaluateCompaction();
            }
        });
    }

    public synchronized void evaluateCompaction() {
        if (compactionPlanner.needsCompaction()) {
            scheduleCompaction();
        }
    }

    public synchronized Future<?> scheduleCompaction() {
        final int id = compactionId.incrementAndGet();
        logger.debug("Starting compaction " + id);

        CompactionPlan compactionPlan = compactionPlanner.planCompaction();

        if (compactionPlan == null) {
            logger.debug("No compaction tasks present " + id);
            logger.debug("Finishing compaction " + id);
            return Futures.immediateFuture(null);
        }

        final List<Future<?>> taskFutures = new ArrayList<Future<?>>();
        Throttle compactionThrottle = new Throttle(config.maxCompactionRate());

        for (CompactionTask task : compactionPlan) {
            logger.debug("Compaction " + id + "  task : " + task);

            for (Table table : task.tables()){
                compactionTables.markAsCompacted(table);
            }

            ThreadPoolExecutor taskExecutor = task.priority().equals(CompactionTask.Priority.HIGH) ?
                    highPriorityCompactionTaskExecutor : compactionTaskExecutor;

            taskFutures.add(taskExecutor.submit(new Task(task, compactionThrottle)));
        }

        metrics.histogram("compactor.concurrentTasks").update(highPriorityCompactionTaskExecutor.getActiveCount() +
                compactionTaskExecutor.getActiveCount());

        return compactionExecutor.submit(new Runnable() {
            @Override
            public void run() {
                for (Future<?> taskFuture : taskFutures){
                    Futures.getUnchecked(taskFuture);
                }

                logger.debug("Finishing compaction " + id);
            }
        });
    }

    public void close() throws IOException {
        compactionExecutor.shutdownNow();
        compactionTaskExecutor.shutdownNow();
    }

    @Override
    public String toString() {
        return "Compactor{" +
                "compactionPlanner=" + compactionPlanner +
                '}';
    }
} 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 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 TableReader implements Iterable<Tuple> {

    private final TableAggregationIterator.Source ascendingIteratorSource = new TableAggregationIterator.Source() {
        @Override
        public CloseableIterator<Tuple> refresh(Key key, long snapshotId) {
            tables.readLock();

            try {
                List<CloseableIterator<Tuple>> tableIterators = new ArrayList<CloseableIterator<Tuple>>();

                for (Table table : tables) {
                    tableIterators.add(key == null ? table.ascendingIterator(snapshotId) : table.ascendingIterator(key,
                            snapshotId));
                }

                return new LatestTupleIterator(snapshotId, new MergingIterator<Tuple>(tableIterators));
            } finally {
                tables.readUnlock();
            }
        }
    };

    private final TableAggregationIterator.Source descendingIteratorSource = new TableAggregationIterator.Source() {
        @Override
        public CloseableIterator<Tuple> refresh(Key key, long snapshotId) {
            tables.readLock();

            try {
                List<CloseableIterator<Tuple>> tableIterators = new ArrayList<CloseableIterator<Tuple>>();

                for (Table table : tables) {
                    tableIterators.add(key == null ? table.descendingIterator(snapshotId) : table.descendingIterator(key,
                            snapshotId));
                }

                return new LatestTupleIterator(snapshotId, new MergingIterator<Tuple>(true, tableIterators));
            } finally {
                tables.readUnlock();
            }
        }
    };

    private final Tables tables;
    private final Metrics metrics;

    private final CacheHitGauge bloomFilterFalsePositiveRate;
    private final Histogram tablesConsultedHistogram;
    private final CacheHitGauge recordNotFoundRate;

    public TableReader(Tables tables, Metrics metrics) {
        this.tables = tables;
        this.metrics = metrics;

        this.bloomFilterFalsePositiveRate = metrics.hitGauge("read.bloomFilterFalsePositiveRate");
        this.tablesConsultedHistogram = metrics.histogram("read.tablesConsulted");
        this.recordNotFoundRate = metrics.hitGauge("read.recordNotFoundRate");
    }

    public Tuple get(Key key) {
        Tuple closestTuple = null;

        tables.readLock();

        int tablesConsulted = 0;

        try {
            for (Table table : tables) {
                if (table.mightContain(key)) {
                    Tuple tableTuple = table.get(key);
                    tablesConsulted++;

                    bloomFilterFalsePositiveRate.sample(tableTuple == null);

                    if (tableTuple != null) {
                        if (closestTuple == null || tableTuple.key().snapshotId() > closestTuple.key().snapshotId()) {
                            closestTuple = tableTuple;
                        }
                    }
                }
            }
        } finally {
            tables.readUnlock();
        }

        tablesConsultedHistogram.update(tablesConsulted);
        recordNotFoundRate.sample(closestTuple == null);

        return closestTuple;
    }

    public CloseableIterator<Tuple> ascendingIterator(long snapshotId) {
        tables.readLock();

        try {
            List<CloseableIterator<Tuple>> tableIterators = new ArrayList<CloseableIterator<Tuple>>();

            for (Table table : tables) {
                tableIterators.add(table.ascendingIterator(snapshotId));
            }

            TableAggregationIterator tableAggregationIterator = new TableAggregationIterator(new MergingIterator<Tuple>
                    (tableIterators), snapshotId, tables, ascendingIteratorSource);

            return new LatestTupleIterator(snapshotId, tableAggregationIterator);
        } finally {
            tables.readUnlock();
        }
    }

    public CloseableIterator<Tuple> descendingIterator(long snapshotId) {
        tables.readLock();

        try {
            List<CloseableIterator<Tuple>> tableIterators = new ArrayList<CloseableIterator<Tuple>>();

            for (Table table : tables) {
                tableIterators.add(table.descendingIterator(snapshotId));
            }

            TableAggregationIterator tableAggregationIterator = new TableAggregationIterator(new MergingIterator<Tuple>
                    (true, tableIterators), snapshotId, tables, descendingIteratorSource);

            return new LatestTupleIterator(snapshotId, tableAggregationIterator);
        } finally {
            tables.readUnlock();
        }
    }

    public CloseableIterator<Tuple> ascendingIterator(Key key, long snapshotId) {
        tables.readLock();

        try {
            List<CloseableIterator<Tuple>> tableIterators = new ArrayList<CloseableIterator<Tuple>>();

            for (Table table : tables) {
                tableIterators.add(table.ascendingIterator(key, snapshotId));
            }

            TableAggregationIterator tableAggregationIterator = new TableAggregationIterator(new MergingIterator<Tuple>
                    (tableIterators), snapshotId, tables, ascendingIteratorSource);

            return new LatestTupleIterator(snapshotId, tableAggregationIterator);
        } finally {
            tables.readUnlock();
        }
    }

    public CloseableIterator<Tuple> descendingIterator(Key key, long snapshotId) {
        tables.readLock();

        try {
            List<CloseableIterator<Tuple>> tableIterators = new ArrayList<CloseableIterator<Tuple>>();

            for (Table table : tables) {
                tableIterators.add(table.descendingIterator(key, snapshotId));
            }

            TableAggregationIterator tableAggregationIterator = new TableAggregationIterator(new MergingIterator<Tuple>
                    (true, tableIterators), snapshotId, tables, descendingIteratorSource);

            return new LatestTupleIterator(snapshotId, tableAggregationIterator);
        } finally {
            tables.readUnlock();
        }
    }

    public synchronized void close() throws IOException {
        tables.readLock();

        try {
            for (Table table : tables) {
                table.close();
            }
        } finally {
            tables.readUnlock();
        }
    }

    @Override
    public Iterator<Tuple> iterator() {
        return ascendingIterator(Long.MAX_VALUE);
    }
} public class Caches {

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

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

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

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

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

    private final Path logDirectory;
    private final Path tableDirectory;

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

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

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

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

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

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

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

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

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

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

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

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

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

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

        filePaths.close();

        return sortedFilePaths;
    }

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

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

        return sortedFileIds;
    }

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

    private final AtomicLong currentSnapshotId = new AtomicLong();
    private final SortedSet<Long> retainedSnapshots = new TreeSet<Long>();

    public Snapshots(long startingSnapshotId) {
        this.currentSnapshotId.set(startingSnapshotId);
        retainedSnapshots.add(Long.MAX_VALUE);
    }

    public long nextId() {
        return currentSnapshotId.incrementAndGet();
    }

    public long currentId() {
        return currentSnapshotId.get();
    }

    public long minimumRetainedId(){
        return retainedSnapshots.first();
    }

    public synchronized void retain(long snapshotId){
        retainedSnapshots.add(snapshotId);
    }

    public synchronized void release(long snapshotId){
        retainedSnapshots.remove(snapshotId);
    }
} public class Tables implements Iterable<Table> {

    public interface ChangeHandler {
        public void changed();
    }

    private final AtomicLong currentTableId = new AtomicLong();
    private final NavigableSet<Table> tables = new TreeSet<Table>();
    private final ReadWriteLock tableLock = new ReentrantReadWriteLock();
    private final List<ChangeHandler> changeHandlers = new ArrayList<ChangeHandler>();

    public Tables(Collection<Table> initialTables) {
        this.tables.addAll(initialTables);
        this.currentTableId.set(tables.isEmpty() ? 0 : tables.last().id());
    }

    public synchronized void addChangeHandler(ChangeHandler changeHandler) {
        changeHandlers.add(changeHandler);
    }

    public synchronized void removeChangeHandler(ChangeHandler changeHandler) {
        changeHandlers.remove(changeHandler);
    }

    public long nextId() {
        return currentTableId.incrementAndGet();
    }

    public long currentId() {
        return currentTableId.get();
    }

    public void readLock() {
        tableLock.readLock().lock();
    }

    public void readUnlock() {
        tableLock.readLock().unlock();
    }

    public void add(Table toAdd) {
        try {
            tableLock.writeLock().lock();
            tables.add(toAdd);
            notifyChanged();
        } finally {
            tableLock.writeLock().unlock();
        }
    }

    public void removeAll(List<Table> toRemove) {
        try {
            tableLock.writeLock().lock();
            for (Table table : toRemove) {
                tables.remove(table);
            }
            notifyChanged();
        } finally {
            tableLock.writeLock().unlock();
        }
    }

    public void remove(Table toRemove) {
        try {
            tableLock.writeLock().lock();
            tables.remove(toRemove);
            notifyChanged();
        } finally {
            tableLock.writeLock().unlock();
        }
    }

    public void swap(Table toAdd, Table toRemove) {
        try {
            tableLock.writeLock().lock();
            tables.remove(toRemove);
            tables.add(toAdd);
            notifyChanged();
        } finally {
            tableLock.writeLock().unlock();
        }
    }

    public int count() {
        return tables.size();
    }

    @Override
    public Iterator<Table> iterator() {
        return tables.iterator();
    }

    private synchronized void notifyChanged() {
        for (ChangeHandler changeHandler : changeHandlers) {
            changeHandler.changed();
        }
    }
} public interface CloseableIterator<T> extends Iterator<T>, Closeable {

    public static class Wrapper<T> implements CloseableIterator<T> {

        private final Iterator<T> delegate;

        public Wrapper(Iterator<T> delegate) {
            this.delegate = delegate;
        }

        @Override
        public void close() throws IOException {

        }

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

        @Override
        public T next() {
            return delegate.next();
        }

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

} public class TableWriter {

    private static final Logger logger = LoggerFactory.getLogger(TableWriter.class);

    private final Config config;
    private final Snapshots snapshots;
    private final ThreadPoolExecutor tableExecutor;
    private final Tables tables;
    private final Paths paths;
    private final Caches caches;
    private final Metrics metrics;
    private final Throttle writeThrottle;

    private MemoryTable memoryTable;
    private CommitLogWriter commitLogWriter;

    public TableWriter(Config config, Paths paths, Tables tables, Snapshots snapshots, Caches caches, Metrics metrics) {
        this.config = config;
        this.paths = paths;
        this.tables = tables;
        this.snapshots = snapshots;
        this.caches = caches;
        this.metrics = metrics;
        this.writeThrottle = new Throttle(config.maxWriteRate());

        this.tableExecutor = new ThreadPoolExecutor(config.tableWriterThreads(), config.tableWriterThreads(),
                Long.MAX_VALUE, TimeUnit.DAYS, new LinkedBlockingQueue<Runnable>(config.tableWriterThreads()),
                new ThreadFactoryBuilder().setNameFormat("Table writer thread %d").build(),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    public synchronized Snapshot write(ByteBuffer key, ByteBuffer value, boolean fsync) throws IOException {
        if (memoryTable == null || memoryTable.size() >= config.memoryTableSize()) {
            rotateMemoryTable();
        }

        int valueCapacity = value == null ? 0 : value.capacity();
        writeThrottle.consume(key.capacity() + valueCapacity);
        long nextSnapshotId = snapshots.nextId();

        key.rewind();

        if (value != null) {
            value.rewind();
        }

        Key recordKey = new Key(key, nextSnapshotId);
        Value recordValue = value == null ? Value.TOMBSTONE_VALUE : new Value(value);
        Tuple tuple = new Tuple(recordKey, recordValue);

        commitLogWriter.append(tuple, fsync);
        memoryTable.put(tuple);

        return new Snapshot(nextSnapshotId);
    }

    public void close() throws IOException {
        if (memoryTable != null) {
            commitLogWriter.close();
        }

        tableExecutor.shutdownNow();
    }

    private void rotateMemoryTable() throws IOException {
        if (memoryTable != null) {
            commitLogWriter.close();
            writeMemoryTable(memoryTable);
        }

        long nextTableId = tables.nextId();
        memoryTable = new MemoryTable(nextTableId);
        commitLogWriter = CommitLogWriter.open(nextTableId, paths);
        tables.add(memoryTable);
    }

    private void writeMemoryTable(final Table tableToWrite) {
        final FileTableWriter.Task task = new FileTableWriter.Task.Builder().tableId(tableToWrite.id()).level(1)
                .paths(paths).config(config).source(tableToWrite.ascendingIterator(snapshots.minimumRetainedId())).tupleCount
                        (tableToWrite.tupleCount()).throttle(Throttle.MAX).callback(new FileTableWriter.Task.Callback
                        () {
            @Override
            public void finish() {
                try {
                    tables.swap(FileTable.open(tableToWrite.id(), paths, caches.recordBlockCache(),
                            caches.indexBlockCache(), metrics), tableToWrite);
                    Files.deleteIfExists(paths.logPath(tableToWrite.id()));
                } catch (ClosedChannelException e) {
                    logger.debug("File table was only partially written " + tableToWrite.id());
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }).build();

        tableExecutor.execute(new Runnable() {
            @Override
            public void run() {
                Timer.Context watch = metrics.timer("write.memoryTableSerialize").time();
                task.run();
                watch.stop();
            }
        });

        metrics.histogram("write.concurrentMemoryTableSerializers").update(tableExecutor.getActiveCount());
    }
}
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.Future;
import com.codahale.metrics.Meter;
import com.codahale.metrics.Timer;
import com.jordanwilliams.heftydb.compact.Compactor;
import com.jordanwilliams.heftydb.data.Key;
import com.jordanwilliams.heftydb.data.Tuple;
import com.jordanwilliams.heftydb.metrics.Metrics;
import com.jordanwilliams.heftydb.read.TableReader;
import com.jordanwilliams.heftydb.state.Caches;
import com.jordanwilliams.heftydb.state.Paths;
import com.jordanwilliams.heftydb.state.Snapshots;
import com.jordanwilliams.heftydb.state.Tables;
import com.jordanwilliams.heftydb.util.CloseableIterator;
import com.jordanwilliams.heftydb.write.TableWriter;
/*
 * Copyright (c) 2014. Jordan Williams
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.jordanwilliams.heftydb.db;



/**
 * The main database API implementation
 */
public class HeftyDB implements DB {

    private class InstrumentedScanIterator implements CloseableIterator<Record> {

        private final CloseableIterator<Record> delegate;

        private InstrumentedScanIterator(CloseableIterator<Record> delegate) {
            this.delegate = delegate;
        }

        @Override
        public boolean hasNext() {
            Timer.Context watch = scanTimer.time();
            boolean hasNext = delegate.hasNext();
            watch.stop();
            return hasNext;
        }

        @Override
        public Record next() {
            Record record = delegate.next();
            scanRate.mark(record.key().capacity() + record.value().capacity());
            return record;
        }

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

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

    private final TableWriter tableWriter;