public class AsyncException extends Exception {
    private static final long serialVersionUID = 1L;

    public AsyncException(Throwable t) {
        super(t);
    }

    public AsyncException(String msg) {
        super(msg);
    }
} public class Batch<T extends Op> {
    private Map<Address, T> map;

    /**
     * Normally you should not create your own Batches but instead have {@link DataCubeIo} create
     * them for you. You can use this if you intend to bypass the high-level magic and you really
     * know what you're doing.
     */
    public Batch() {
        this.map = Maps.newHashMap();
    }

    /**
     * Normally you should not create your own Batches but instead have {@link DataCubeIo} create
     * them for you. You can use this if you intend to bypass the high-level magic and you really
     * know what you're doing.
     *
     * @param map some Ops to wrap in this Batch
     */
    public Batch(Map<Address, T> map) {
        this.map = map;
    }

    public void putAll(Batch<T> b) {
        this.putAll(b.getMap());
    }

    @SuppressWarnings("unchecked")
    public void putAll(Map<Address, T> other) {
        for (Map.Entry<Address, T> entry : other.entrySet()) {
            Address c = entry.getKey();
            T alreadyExistingVal = map.get(entry.getKey());
            T newVal;
            if (alreadyExistingVal == null) {
                newVal = entry.getValue();
            } else {
                newVal = (T) alreadyExistingVal.add(entry.getValue());
            }
            this.map.put(c, newVal);
        }
    }

    public Map<Address, T> getMap() {
        return map;
    }

    public String toString() {
        return map.toString();
    }

    public void reset() {
        map = Maps.newHashMap();
    }
} public class BucketType {
    public static final BucketType IDENTITY = new BucketType("nobucket", new byte[]{});
    public static final BucketType WILDCARD = new BucketType("wildcard", new byte[]{});

    private final String nameInErrMsgs;
    private final byte[] uniqueId;

    public BucketType(String nameInErrMsgs, byte[] uniqueId) {
        this.nameInErrMsgs = nameInErrMsgs;
        this.uniqueId = uniqueId;
    }

    public byte[] getUniqueId() {
        return uniqueId;
    }

    public String getNameInErrMsgs() {
        return nameInErrMsgs;
    }

    public String toString() {
        return nameInErrMsgs;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (!(o instanceof BucketType)) return false;
        BucketType that = (BucketType) o;
        return Objects.equals(nameInErrMsgs, that.nameInErrMsgs) &&
                Arrays.equals(uniqueId, that.uniqueId);
    }

    @Override
    public int hashCode() {

        int result = Objects.hash(nameInErrMsgs);
        result = 31 * result + Arrays.hashCode(uniqueId);
        return result;
    }
} public class DataCube<T extends Op> {
    private static final Logger log = LoggerFactory.getLogger(DataCube.class);

    public enum PREFIX_MODE {NO_ADDRESS_PREFIX, MOD_ADDRESS_PREFIX}

    private final List<Dimension<?>> dims;
    private final List<Rollup> rollups;
    private final Multimap<Dimension<?>, BucketType> bucketsOfInterest;
    private final Map<Set<DimensionAndBucketType>, Rollup> validAddressSet;
    private final boolean useAddressPrefixByteHash;

    /**
     * @param dims    see {@link Dimension}
     * @param rollups see {@link Rollup}
     */
    public DataCube(List<Dimension<?>> dims, List<Rollup> rollups) {
        this(dims, rollups, PREFIX_MODE.NO_ADDRESS_PREFIX);
    }

    /**
     * @param dims       see {@link Dimension}
     * @param rollups    see {@link Rollup}
     * @param prefixMode use MOD_ADDRESS_PREFIX to prefix the keys by a hash byte (calculated by hashing each element
     *                   in the key).  This is only a storage artifact to benefit systems like HBase, where
     *                   monotonically increasing row keys can result in hot spots.
     *
     *                   Warning: Do NOT switch modes for an existing cube or the keys will
     *                   not map properly.  Also, data from versions of datacube before 2.0.0,
     *                   with this feature enabled, is not compatible with 2.0.0+.
     */
    public DataCube(List<Dimension<?>> dims, List<Rollup> rollups, PREFIX_MODE prefixMode) {
        this.dims = dims;
        this.rollups = ImmutableList.copyOf(rollups);
        this.validAddressSet = Maps.newHashMap();
        if (PREFIX_MODE.MOD_ADDRESS_PREFIX == prefixMode) {
            this.useAddressPrefixByteHash = true;
        } else {
            this.useAddressPrefixByteHash = false;
        }

        bucketsOfInterest = HashMultimap.create();

        if (dims.size() > Short.MAX_VALUE) {
            throw new IllegalArgumentException("May not have more than " + Short.MAX_VALUE +
                    " dimensions");
        }

        for (Rollup rollup : rollups) {
            for (DimensionAndBucketType dimAndBucketType : rollup.getComponents()) {
                if (!dims.contains(dimAndBucketType.dimension)) {
                    throw new IllegalArgumentException("Rollup dimension " +
                            dimAndBucketType.dimension + " is not a dimension in this cube");
                }

                bucketsOfInterest.put(dimAndBucketType.dimension, dimAndBucketType.bucketType);
            }
            validAddressSet.put(new HashSet<>(rollup.getComponents()), rollup);
        }
    }

    /**
     * Get a batch of writes that, when applied to the database, will make the change given by
     * "op".
     */
    public Batch<T> getWrites(WriteBuilder writeBuilder, T op) {
        Map<Address, T> outputMap = Maps.newHashMap();

        for (Rollup rollup : rollups) {

            List<Set<byte[]>> coordSets =
                    new ArrayList<Set<byte[]>>(rollup.getComponents().size());

            boolean dimensionHadNoBucket = false;
            for (DimensionAndBucketType dimAndBucketType : rollup.getComponents()) {
                Dimension<?> dimension = dimAndBucketType.dimension;
                BucketType bucketType = dimAndBucketType.bucketType;

                SetMultimap<BucketType, byte[]> bucketsForDimension =
                        writeBuilder.getBuckets().get(dimension);

                if (bucketsForDimension == null || bucketsForDimension.isEmpty()) {
                    dimensionHadNoBucket = true;
                    break;
                }

                Set<byte[]> coords = bucketsForDimension.get(bucketType);
                if (coords == null || coords.isEmpty()) {
                    dimensionHadNoBucket = true;
                    break;
                }
                coordSets.add(coords);
            }

            if (dimensionHadNoBucket) {
                // Skip this rollup since one of its input buckets was not present.
                log.debug("Skipping rollup due to dimension with no buckets");
                continue;
            }

            Set<List<byte[]>> crossProduct = Sets.cartesianProduct(coordSets);

            for (List<byte[]> crossProductTuple : crossProduct) {
                Address outputAddress = new Address(this, rollup);

                // Start out with all dimensions wildcard, overwrite with other values later
                for (Dimension<?> dimension : dims) {
                    outputAddress.at(dimension, BucketTypeAndBucket.WILDCARD);
                }

                for (int i = 0; i < crossProductTuple.size(); i++) {
                    byte[] coord = crossProductTuple.get(i);
                    Dimension<?> dim = rollup.getComponents().get(i).dimension;
                    BucketType bucketType = rollup.getComponents().get(i).bucketType;

                    outputAddress.at(dim, new BucketTypeAndBucket(bucketType, coord));
                }

                boolean shouldWrite = true;

                if (shouldWrite) {
                    outputMap.put(outputAddress, op);
                }
            }
        }

        return new Batch<T>(outputMap);
    }

    public boolean useAddressPrefixByteHash() {
        return useAddressPrefixByteHash;
    }

    public List<Dimension<?>> getDimensions() {
        return dims;
    }

    /**
     * @throws IllegalArgumentException if addr isn't in the cube.
     */
    void checkValidReadOrThrow(Address addr) {
        Set<DimensionAndBucketType> dimsAndBucketsSpecified = new HashSet<DimensionAndBucketType>(addr.getBuckets().size());

        // Find out which dimensions have literal values (not wildcards)
        for (Entry<Dimension<?>, BucketTypeAndBucket> e : addr.getBuckets().entrySet()) {
            BucketTypeAndBucket bucketTypeAndCoord = e.getValue();
            if (bucketTypeAndCoord == BucketTypeAndBucket.WILDCARD) {
                continue;
            }

            Dimension<?> dimension = e.getKey();
            BucketType bucketType = bucketTypeAndCoord.bucketType;

            dimsAndBucketsSpecified.add(new DimensionAndBucketType(dimension, bucketType));
        }

        // Make sure that the requested address exists in this cube (is touched by some rollup)
        final Rollup sourceRollup = validAddressSet.get(dimsAndBucketsSpecified);
        if (sourceRollup != null) {
            addr.setSourceRollup(sourceRollup);
            return;
        }

        // The value the user asked for isn't written by this cube
        StringBuilder errMsg = new StringBuilder();
        errMsg.append("Won't query with the dimensions given, since there are no rollups that store that ");
        errMsg.append("value. You gave dimensions: ");
        errMsg.append(dimsAndBucketsSpecified);
        errMsg.append(". To query, you must provide one of these complete sets of dimensions and buckets: (");
        boolean firstLoop = true;
        for (Rollup rollup : rollups) {
            if (!firstLoop) {
                errMsg.append(",");
            }
            firstLoop = false;
            errMsg.append(rollup.getComponents());
        }
        errMsg.append(")");
        throw new IllegalArgumentException(errMsg.toString());
    }

    Multimap<Dimension<?>, BucketType> getBucketsOfInterest() {
        return bucketsOfInterest;
    }

    public List<Rollup> getRollups() {
        return rollups;
    }
} public class DataCubeIo<T extends Op> {
    private static final Logger log = LoggerFactory.getLogger(DataCubeIo.class);

    private final DbHarness<T> db;
    private final DataCube<T> cube;
    private final int batchSize;
    private final long maxBatchAgeMs;
    private final SyncLevel syncLevel;
    private AsyncException asyncException = null;

    private final Object lock = new Object();

    // This executor will wait for DB writes to complete then check if they had an error.
    private final ThreadPoolExecutor asyncErrorMonitorExecutor;

    private final String metricsScope;
    private final Meter writesMeter;
    private final Meter asyncQueueBackoffMeter;
    private final Meter runBatchMeter;
    private final Meter ageFlushes;
    private final Meter sizeFlushes;

    private final boolean perRollupMetrics;
    private final Counter writeAddrWithoutRollup;
    private final ConcurrentMap<Rollup, Counter> rollupReadNotFound = Maps.newConcurrentMap();
    private final ConcurrentMap<Rollup, Counter> rollupReadCount = Maps.newConcurrentMap();
    private final ConcurrentMap<Rollup, Counter> rollupWriteIndividual = Maps.newConcurrentMap();
    private final ConcurrentMap<Rollup, Histogram> rollupReadSize = Maps.newConcurrentMap();
    private final ConcurrentMap<Rollup, Histogram> rollupWriteSize = Maps.newConcurrentMap();

    private Batch<T> batchInProgress = new Batch<T>();
    private long batchFlushDeadlineMs;

    /**
     * @param batchSize     if after doing a write the number of rows to be written to the database
     *                      exceeds this number, a flush will be done.
     * @param maxBatchAgeMs if after doing a write the batch's oldest write was more than this long ago,
     *                      a flush will be done. This is not a hard ceiling on the age of writes in the batch, because
     *                      the batch will not be flushed until the *next* write arrives after the timeout.
     */
    public DataCubeIo(DataCube<T> cube, DbHarness<T> db, int batchSize, long maxBatchAgeMs,
                      SyncLevel syncLevel, String metricsScope, boolean perRollupMetrics) {
        this.cube = cube;
        this.db = db;
        this.batchSize = batchSize;
        this.maxBatchAgeMs = maxBatchAgeMs;
        this.syncLevel = syncLevel;

        this.perRollupMetrics = perRollupMetrics;
        this.metricsScope = metricsScope;
        writesMeter = Metrics.meter(DataCubeIo.class, "writes", metricsScope);
        asyncQueueBackoffMeter = Metrics.meter(DataCubeIo.class, "backoffMeter", metricsScope);
        runBatchMeter = Metrics.meter(DataCubeIo.class, "runBatchMeter", metricsScope);
        ageFlushes = Metrics.meter(DataCubeIo.class, "flushesDueToAge", metricsScope);
        sizeFlushes = Metrics.meter(DataCubeIo.class, "flushesDueToSize", metricsScope);
        writeAddrWithoutRollup = Metrics.counter(DataCubeIo.class, "writeAddrWithoutRollup", metricsScope);

        this.asyncErrorMonitorExecutor = new ThreadPoolExecutor(Integer.MAX_VALUE,
                Integer.MAX_VALUE, 1, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(),
                new NamedThreadFactory("DataCubeIo async DB watcher"));

        if (perRollupMetrics) {
            cube.getRollups().forEach(rollup -> rollupReadCount.put(rollup, Metrics.counter(
                    DataCubeIo.class, metricsScope + "_rollupReadCount", rollup.getMetricName()
            )));
        }

        Metrics.gauge(DataCubeIo.class, "errorMonitorActiveCount", metricsScope, new Gauge<Integer>() {
            @Override
            public Integer getValue() {
                return asyncErrorMonitorExecutor.getActiveCount();
            }
        });

    }

    /**
     * Do some writes into the in-memory batch, possibly flushing to the backing database, depending
     * on the {@link SyncLevel}.
     *
     * @return If the operation did not cause a database flush, Optional.absent will be returned.
     * Otherwise, if a batch flush is triggered, a Future will be returned. This Future's get()
     * method will return null when the database flush is finished, and rethrow any exceptions as
     * ExecutionExceptions (see {@link java.util.concurrent.Future}).
     */
    public Optional<Future<?>> writeAsync(T op, WriteBuilder at) throws AsyncException, InterruptedException {
        if (asyncException != null) {
            throw asyncException;
        }

        Batch<T> newBatch = cube.getWrites(at, op);

        return writeAsync(newBatch);
    }

    public Optional<Future<?>> writeAsync(Batch<T> newBatch) throws AsyncException, InterruptedException {
        writesMeter.mark();

        Batch<T> batchToFlush = null;

        if (perRollupMetrics) {
            newBatch.getMap().keySet().forEach(addr -> {
                if (addr.getSourceRollup().isPresent()) {
                    updateRollupCounter(rollupWriteIndividual, addr.getSourceRollup().get(), "rollupIndividualWrite");
                } else {
                    writeAddrWithoutRollup.inc();
                }
            });
        }

        switch (syncLevel) {
            case FULL_SYNC:
                // At this sync level we flush all batches immediately. No sharing between threads.
                batchToFlush = newBatch;
                break;
            case BATCH_ASYNC:
            case BATCH_SYNC:
                synchronized (lock) {
                    if (batchInProgress.getMap().isEmpty()) {
                        // Start the timer for this batch, it should be flushed when it becomes old
                        long nowTimeMs = System.currentTimeMillis();
                        batchFlushDeadlineMs = nowTimeMs + maxBatchAgeMs; // Flush when we reach this timestamp

                        // If the flush deadline timestamp overflowed its long, set it back to the largest
                        // possible value. This will occur if the client passes a very large max batch age.
                        if (batchFlushDeadlineMs < nowTimeMs) {
                            batchFlushDeadlineMs = Long.MAX_VALUE;
                        }
                    }
                    batchInProgress.putAll(newBatch);

                    boolean shouldFlush = false;

                    if (batchInProgress.getMap().size() >= batchSize) {
                        sizeFlushes.mark();
                        shouldFlush = true;
                    } else if (System.currentTimeMillis() >= batchFlushDeadlineMs) {
                        ageFlushes.mark();
                        shouldFlush = true;
                    }

                    if (shouldFlush) {
                        batchToFlush = batchInProgress;
                        batchInProgress = new Batch<T>();
                    }
                }
                break;
            default:
                throw new RuntimeException("Unknown sync level " + syncLevel);

        }

        if (batchToFlush != null) {
            return Optional.<Future<?>>of(runBatch(batchToFlush));
        } else {
            return Optional.empty();
        }
    }

    /**
     * Hand off a batch to the DbHarness layer, retrying on FullQueueException.
     */
    private Future<?> runBatch(Batch<T> batch) throws InterruptedException {
        while (true) {
            try {
                runBatchMeter.mark();
                if (perRollupMetrics) {
                    batch.getMap().forEach((addr, op) -> {
                        if (addr.getSourceRollup().isPresent()) {
                            updateRollupHistogram(rollupWriteSize, addr.getSourceRollup().get(), "rollupWriteSize", op);
                        }
                    });
                }

                return db.runBatchAsync(batch, flushErrorHandler);
            } catch (FullQueueException e) {
                asyncQueueBackoffMeter.mark();

                // Sleeping and retrying like this means batches may be flushed out of order
                log.debug("Async queue is full, retrying soon");
                Thread.sleep(100);
            }
        }
    }

    /**
     * If the {@link SyncLevel} is {@link SyncLevel#FULL_SYNC} or {@link SyncLevel#BATCH_SYNC}, then
     * no asynchronous IO is happening. You can use this function to write instead of
     * {@link #writeAsync(Op, WriteBuilder)} without catching AsyncException or InterruptedException.
     * <p>
     * IMPORTANT: the name of this function does not imply that the write is immediately flushed to
     * the database. This is only true if {@link SyncLevel#FULL_SYNC} is set. Otherwise your write
     * is probably just staged in a batch for writing later.
     * <p>
     * You can only use this function if this DataCubeIo was constructed with
     * {@link SyncLevel#FULL_SYNC} or {@link SyncLevel#BATCH_SYNC}.
     */
    public void writeSync(T op, WriteBuilder at) throws IOException, InterruptedException {
        if (syncLevel == SyncLevel.BATCH_ASYNC) {
            throw new IllegalArgumentException("You can't use WriteSync for this cube with " +
                    "SyncLevel " + syncLevel);
        }

        Optional<Future<?>> optFuture;
        try {
            optFuture = writeAsync(op, at);
        } catch (AsyncException pe) {
            throw new RuntimeException("Internal error, when at a synchronized syncLevel there should" +
                    " be no asynchronous exceptions");
        }

        if (optFuture.isPresent()) {
            // Our write triggered a batch flush. Wait for it to finish, rethrowing exceptions.
            try {
                optFuture.get().get();
            } catch (ExecutionException ee) {
                Throwable flushException = ee.getCause();
                if (flushException instanceof IOException) {
                    throw (IOException) flushException;
                } else if (flushException instanceof InterruptedException) {
                    throw (InterruptedException) flushException;
                } else if (flushException instanceof RuntimeException) {
                    throw new RuntimeException(flushException);
                } else {
                    throw new RuntimeException("Unreachable");
                }
            }
        } else {
            // Our write updated a batch in memory without causing a flush. Return success.
        }
    }

    /**
     * @return absent if the bucket doesn't exist, or the bucket if it does.
     */
    public Optional<T> get(Address addr) throws IOException, InterruptedException {
        cube.checkValidReadOrThrow(addr);

        final Optional<T> result = db.get(addr);
        if (perRollupMetrics && addr.getSourceRollup().isPresent()) {
            updateRollupCounter(rollupReadCount, addr.getSourceRollup().get(), "rollupReadCount");
            if (result.isPresent()) {
                updateRollupHistogram(rollupReadSize, addr.getSourceRollup().get(), "rollupReadSize", result.get());
            } else {
                updateRollupCounter(rollupReadNotFound, addr.getSourceRollup().get(), "rollupReadNotFound");
            }
        }

        return result;
    }

    public Optional<T> get(ReadBuilder readBuilder) throws IOException, InterruptedException {
        return this.get(readBuilder.build());
    }

    public List<Optional<T>> multiGet(List<ReadBuilder> readBuilders) throws IOException, InterruptedException {
        List<Address> addresses = Lists.newArrayListWithCapacity(readBuilders.size());
        for (ReadBuilder readBuilder : readBuilders) {
            Address address = readBuilder.build();
            cube.checkValidReadOrThrow(address);
            addresses.add(address);
        }

        final List<Optional<T>> results = db.multiGet(addresses);
        if (perRollupMetrics) {
            for (int i = 0; i < results.size(); i++) {
                final Optional<T> result = results.get(i);
                final Address addr = addresses.get(i);
                if (!addr.getSourceRollup().isPresent()) {
                    //Should never happen, but be paranoid
                    continue;
                }

                updateRollupCounter(rollupReadCount, addr.getSourceRollup().get(), "rollupReadCount");
                if (result.isPresent()) {
                    updateRollupHistogram(rollupReadSize, addr.getSourceRollup().get(), "rollupReadSize", result.get());
                } else {
                    updateRollupCounter(rollupReadNotFound, addr.getSourceRollup().get(), "rollupReadNotFound");
                }
            }
        }

        return results;
    }

    public void flush() throws InterruptedException {
        Batch<T> batchToFlush;
        synchronized (lock) {
            batchToFlush = batchInProgress;
            batchInProgress = new Batch<T>();
        }
        runBatch(batchToFlush);
        db.flush();
    }

    private AfterExecute<T> flushErrorHandler = new AfterExecute<T>() {
        @Override
        public void afterExecute(Throwable t) {
            if (t != null) {
                asyncException = new AsyncException(t);
                log.error("Putting DataCubeIo into an error state due to flush exception", t);
            }
        }
    };

    private void updateRollupCounter(Map<Rollup, Counter> counters, Rollup rollup, String metric) {
        if (!perRollupMetrics) {
            return;
        }

        final Counter counter = counters.computeIfAbsent(rollup, ru ->
                Metrics.counter(DataCubeIo.class, metricsScope + "_" + metric, rollup.getMetricName()));
        counter.inc();
    }

    private void updateRollupHistogram(Map<Rollup, Histogram> histograms, Rollup rollup, String metric, T op) {
        if (!perRollupMetrics) {
            return;
        }

        final long opNumericValue;
        if (op instanceof LongOp) {
            opNumericValue = ((LongOp) op).getLong();
        } else if (op instanceof IntOp) {
            opNumericValue = ((IntOp) op).getInt();
        } else {
            return;
        }

        final Histogram histogram = histograms.computeIfAbsent(rollup, ru ->
                Metrics.histogram(DataCubeIo.class, metricsScope + "_" + metric, rollup.getMetricName()));
        histogram.update(opNumericValue);
    }
} public interface DbHarness<T extends Op> {
    /**
     * Apply some ops to the database, combining them with ops that might already be stored there.
     * The passed batch will be modified by runBatch() to remove all ops that were successfully
     * stored, so the batch will be empty when this function returns if there was no IOException.
     * If this function throws IOException, the batch will contain one or more ops that were not
     * successfully applied to the DB.
     *
     * This is asynchronous, meaning that the given batch will probably be executed soon, but not
     * necessarily by the time this function returns.
     *
     * @throws AsyncException     if the DbHarness is in a bad state as a result of an earlier
     *                            RuntimeException and will not accept any more batches. See {@link AsyncException} for
     *                            more info.
     * @throws FullQueueException if the queue of pending batches to be flushed is full, and no more
     *                            batches can be accepted right now. The caller can retry soon.
     */
    Future<?> runBatchAsync(Batch<T> batch, AfterExecute<T> afterExecute) throws FullQueueException;

    /**
     * @return absent if the bucket doesn't exist, or the bucket if it does.
     *
     * @throws InterruptedException
     */
    Optional<T> get(Address c) throws IOException, InterruptedException;

    /**
     * Overwrite value.
     *
     * @throws IOException
     * @throws InterruptedException
     */
    void set(Address c, T op) throws IOException, InterruptedException;

    List<Optional<T>> multiGet(List<Address> addresses) throws IOException, InterruptedException;

    /**
     * A hook for shutting down any internal threads during orderly shutdown. The default implementation simply calls
     * {@link #flush}
     *
     * @throws Exception
     */
    default void shutdown() throws IOException, InterruptedException {
        flush();
    }

    /**
     * When it's time to write a batch to the database, there are a couple of ways it can
     * be done.
     *
     * READ_COMBINE_CAS: read the existing cell value from the database, deserialize
     * it into an Op, combine it with the Op to be written, and write the combined Op back
     * to the database using a compare-and-swap operation to make sure it hasn't changed
     * since we read it. If the CAS fails because the cell value was concurrently modifed by
     * someone else, start over at the beginning and re-read the value.
     *
     * INCREMENT: for numeric values like counters, if the database supports server side
     * incrementing, the serialized Op will be passed as the amount to be incremented. This
     * means that no reads or combines will be done in the client (the harness code).
     *
     * OVERWRITE: without reading from the database, overwrite the cell value with the value
     * from the batch in memory. This will wipe out the existing count, so don't use this
     * unless you're sure you know what you're doing.
     */
    enum CommitType {
        READ_COMBINE_CAS,
        INCREMENT,
        OVERWRITE
    }

    /**
     * Wait until all currently outstanding writes have been written to the DB. or failed.
     */
    void flush() throws InterruptedException;
} public class Dimension<F> {
    private final String name;
    private final Bucketer<F> bucketer;
    private final boolean doIdSubstitution;
    private final int numFieldBytes;
    private final int bucketPrefixSize;
    private final boolean isBucketed;
    private final boolean nullable;

    /**
     * For dimensions where a single input bucket affects multiple buckets within that dimension.
     * For example, a single input data point might affect hourly, daily, and monthly counts.
     *
     * @param doIdSubstitution whether to use the immutable bucket->uniqueId substition service
     * @param fieldBytes       the number of bytes that will be reserved in database keys for coordinates in this
     *                         dimension. For example, if values in this dimension have 1000 distinct values, then
     *                         you'd need ceil(log2(1000)/8)=2 bytes (1000 values can be enumerated in 10 bits, which
     *                         requires 2 full bytes).
     */
    public Dimension(String name, Bucketer<F> bucketer, boolean doIdSubstitution, int fieldBytes) {
        this(name, bucketer, doIdSubstitution, fieldBytes, false);
    }

    /**
     * For dimensions where a single input bucket affects multiple buckets within that dimension.
     * For example, a single input data point might affect hourly, daily, and monthly counts.
     *
     * @param doIdSubstitution whether to use the immutable bucket->uniqueId substition service
     * @param fieldBytes       number of bytes that will be reserved in database keys for coordinates in this
     *                         dimension. For example, if values in this dimension have 1000 distinct values, then
     *                         you'd need ceil(log2(1000)/8)=2 bytes (1000 values can be enumerated in 10 bits, which
     *                         requires 2 full bytes).
     * @param nullable         false if input values are required to specify a value for this dimension,
     *                         otherwise true
     */
    public Dimension(String name, Bucketer<F> bucketer, boolean doIdSubstitution, int fieldBytes, boolean nullable) {
        this.name = name;
        this.bucketer = bucketer;
        this.doIdSubstitution = doIdSubstitution;
        this.numFieldBytes = fieldBytes;
        this.nullable = nullable;

        // Make sure all bucket unique id prefixes have the same length
        Integer previousLen = null;
        List<BucketType> bucketTypes = bucketer.getBucketTypes();
        if (bucketTypes.size() == 0) {
            throw new IllegalArgumentException("Invalid bucketer. There must at least one bucket type");
        } else if (bucketTypes.size() == 1 || bucketTypes.get(0) == BucketType.IDENTITY) {
            isBucketed = false;
            bucketPrefixSize = 0;
        } else {
            for (BucketType bucketType : bucketTypes) {
                int thisBucketLen = bucketType.getUniqueId().length;
                if (previousLen == null) {
                    previousLen = thisBucketLen;
                } else {
                    if (previousLen != thisBucketLen) {
                        throw new IllegalArgumentException("BucketTypes for dimension " +
                                name + " had different lengths " + previousLen + " and "
                                + thisBucketLen);
                    }
                }
            }
            bucketPrefixSize = previousLen;
            isBucketed = true;
        }
    }

    public String getName() {
        return name;
    }

    public int getNumFieldBytes() {
        return numFieldBytes;
    }

    public String toString() {
        return name;
    }

    Bucketer<F> getBucketer() {
        return bucketer;
    }

    public boolean getDoIdSubstitution() {
        return doIdSubstitution;
    }

    public boolean isNullable() {
        return nullable;
    }

    public int getBucketPrefixSize() {
        return bucketPrefixSize;
    }

    boolean isBucketed() {
        return isBucketed;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (!(o instanceof Dimension)) return false;
        Dimension<?> dimension = (Dimension<?>) o;
        return doIdSubstitution == dimension.doIdSubstitution &&
                numFieldBytes == dimension.numFieldBytes &&
                bucketPrefixSize == dimension.bucketPrefixSize &&
                isBucketed == dimension.isBucketed &&
                nullable == dimension.nullable &&
                Objects.equals(name, dimension.name) &&
                Objects.equals(bucketer, dimension.bucketer);
    }

    @Override
    public int hashCode() {
        return Objects.hash(name, bucketer, doIdSubstitution, numFieldBytes, bucketPrefixSize, isBucketed, nullable);
    }
} public class Rollup {
    private final List<DimensionAndBucketType> components;
    private final String metricName;

    public Rollup(Dimension<?>... dims) {
        this.components = new ArrayList<>(dims.length);
        for (Dimension<?> dim : dims) {
            this.components.add(new DimensionAndBucketType(dim, BucketType.IDENTITY));
        }
        this.metricName = makeMetricName();
    }

    public Rollup(Set<DimensionAndBucketType> components) {
        this.components = new ArrayList<>(components); // defensive copy
        this.metricName = makeMetricName();
    }

    /**
     * Convenient wrapper around {@link #Rollup(Set)} that builds a set for you.
     */
    public Rollup(Dimension<?> d) {
        this(ImmutableSet.of(new DimensionAndBucketType(d, BucketType.IDENTITY)));
    }

    /**
     * Convenient wrapper around {@link #Rollup(Set)} that builds a set for you.
     */
    public Rollup(Dimension<?> d, BucketType bt) {
        this(ImmutableSet.of(new DimensionAndBucketType(d, bt)));
    }

    /**
     * Convenient wrapper around {@link #Rollup(Set)} that builds a set for you.
     */
    public Rollup(Dimension<?> d1, BucketType bt1, Dimension<?> d2, BucketType bt2) {
        this(ImmutableSet.of(new DimensionAndBucketType(d1, bt1),
                new DimensionAndBucketType(d2, bt2)));
    }

    /**
     * Convenient wrapper around {@link #Rollup(Set)} that builds a set for you.
     */
    public Rollup(Dimension<?> d1, BucketType bt1, Dimension<?> d2, BucketType bt2,
                  Dimension<?> d3, BucketType bt3) {
        this(ImmutableSet.of(new DimensionAndBucketType(d1, bt1),
                new DimensionAndBucketType(d2, bt2), new DimensionAndBucketType(d3, bt3)));
    }

    /**
     * Convenient wrapper around {@link #Rollup(Set)} that builds a set for you.
     */
    public Rollup(Dimension<?> d1, Dimension<?> d2) {
        this(ImmutableSet.of(new DimensionAndBucketType(d1, BucketType.IDENTITY),
                new DimensionAndBucketType(d2, BucketType.IDENTITY)));
    }

    /**
     * Convenient wrapper around {@link #Rollup(Set)} that builds a set for you.
     */
    public Rollup(Dimension<?> d1, Dimension<?> d2, BucketType bt2) {
        this(ImmutableSet.of(new DimensionAndBucketType(d1, BucketType.IDENTITY),
                new DimensionAndBucketType(d2, bt2)));
    }

    public List<DimensionAndBucketType> getComponents() {
        return components;
    }

    public String getMetricName() {
        return metricName;
    }

    private String makeMetricName() {
        return components
                .stream()
                .map(dimAndBucket -> {
                    if (dimAndBucket.bucketType == BucketType.IDENTITY || dimAndBucket.bucketType == BucketType.WILDCARD) {
                        return dimAndBucket.dimension.getName();
                    } else {
                        return dimAndBucket.dimension.getName() + "_" + dimAndBucket.bucketType.getNameInErrMsgs();
                    }
                })
                .sorted()
                .collect(Collectors.joining("-"));
    }

    public String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append("(Rollup over ");
        sb.append(components);
        sb.append(")");
        return sb.toString();
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (!(o instanceof Rollup)) return false;
        Rollup rollup = (Rollup) o;
        return Objects.equals(components, rollup.components) &&
                Objects.equals(metricName, rollup.metricName);
    }

    @Override
    public int hashCode() {
        return Objects.hash(components, metricName);
    }
} public enum SyncLevel {FULL_SYNC, BATCH_SYNC, BATCH_ASYNC} public class WriteBuilder {
    private final Map<Dimension<?>, SetMultimap<BucketType, byte[]>> buckets;

    public WriteBuilder() {
        buckets = Maps.newHashMap();
    }

    public <O> WriteBuilder at(Dimension<O> dimension, O coord) {
        Bucketer<O> bucketer = dimension.getBucketer();

        SetMultimap<BucketType, CSerializable> bucketsAndCoords = bucketer.bucketForWrite(coord);
        buckets.put(dimension, serializeCoords(bucketsAndCoords));

        return this;
    }

    Map<Dimension<?>, SetMultimap<BucketType, byte[]>> getBuckets() {
        return buckets;
    }

    private static SetMultimap<BucketType, byte[]> serializeCoords(
            Multimap<BucketType, CSerializable> inputMap) {
        SetMultimap<BucketType, byte[]> outputMap = HashMultimap.create();
        for (Map.Entry<BucketType, CSerializable> e : inputMap.entries()) {
            outputMap.put(e.getKey(), e.getValue().serialize());
        }
        return outputMap;
    }

    public String toString() {
        return buckets.toString();
    }
} public class HourDayMonthBucketer implements Bucketer<DateTime> {
    public static final BucketType hours = new BucketType("hour", new byte[]{1});
    public static final BucketType days = new BucketType("day", new byte[]{2});
    public static final BucketType months = new BucketType("month", new byte[]{3});
    public static final BucketType years = new BucketType("year", new byte[]{4});
    public static final BucketType weeks = new BucketType("weeks", new byte[]{5});

    static final Set<BucketType> ALL_BUCKET_TYPES = ImmutableSet.<BucketType>builder().
            add(hours).add(days).add(months).add(years).add(weeks).build();

    private final List<BucketType> bucketTypes;

    // TODO move this class to TimeBucker and have HourDayMonthBucketer be a subclass that
    // just calls super(ImmutableList.of(hours, days, months))
    public HourDayMonthBucketer() {
        this(ImmutableList.of(hours, days, months));
    }

    public HourDayMonthBucketer(List<BucketType> bucketTypes) {
        this.bucketTypes = ImmutableList.copyOf(bucketTypes); // defensive copy
        for (BucketType bucketType : bucketTypes) {
            if (!ALL_BUCKET_TYPES.contains(bucketType)) {
                throw new IllegalArgumentException("Invalid bucket type " + bucketType +
                        ", expected one of " + ALL_BUCKET_TYPES);
            }
        }
    }

    @Override
    public SetMultimap<BucketType, CSerializable> bucketForWrite(
            DateTime coordinate) {
        ImmutableSetMultimap.Builder<BucketType, CSerializable> builder =
                ImmutableSetMultimap.builder();
        for (BucketType bucketType : bucketTypes) {
            builder.put(bucketType, bucket(coordinate, bucketType));
        }
        return builder.build();
    }

    @Override
    public CSerializable bucketForRead(Object coordinateField, BucketType bucketType) {
        DateTime inputTime = (DateTime) coordinateField;
        return bucket(inputTime, bucketType);
    }

    private CSerializable bucket(DateTime inputTime, BucketType bucketType) {
        DateTime returnVal;
        if (bucketType == hours) {
            returnVal = hourFloor(inputTime);
        } else if (bucketType == days) {
            returnVal = dayFloor(inputTime);
        } else if (bucketType == months) {
            returnVal = monthFloor(inputTime);
        } else if (bucketType == years) {
            returnVal = yearFloor(inputTime);
        } else if (bucketType == weeks) {
            returnVal = weekFloor(inputTime);
        } else {
            throw new RuntimeException("Unexpected bucket type " + bucketType);
        }
        return new LongSerializable(returnVal.getMillis());
    }

    @Override
    public List<BucketType> getBucketTypes() {
        return ImmutableList.of(hours, days, months, years, weeks);
    }

    public static DateTime hourFloor(DateTime dt) {
        return dt.withMillisOfSecond(0).withSecondOfMinute(0).withMinuteOfHour(0);
    }

    public static DateTime dayFloor(DateTime dt) {
        return hourFloor(dt).withHourOfDay(0);
    }

    public static DateTime monthFloor(DateTime dt) {
        return dayFloor(dt).withDayOfMonth(1);
    }

    public static DateTime yearFloor(DateTime dt) {
        return monthFloor(dt).withMonthOfYear(1);
    }

    public static DateTime weekFloor(DateTime dt) {
        return dayFloor(dt.minusDays(dt.getDayOfWeek() - 1));
    }
} public class StringToBytesBucketer extends AbstractIdentityBucketer<String> {
    private static final StringToBytesBucketer instance = new StringToBytesBucketer();

    public static final StringToBytesBucketer getInstance() {
        return instance;
    }

    @Override
    public CSerializable makeSerializable(String coord) {
        return new StringSerializable(coord);
    }
} public class TagsBucketer implements Bucketer<Collection<String>> {

    @Override
    public SetMultimap<BucketType, CSerializable> bucketForWrite(
            Collection<String> tags) {
        ImmutableSetMultimap.Builder<BucketType, CSerializable> builder =
                ImmutableSetMultimap.builder();
        for (String tag : tags) {
            builder.put(BucketType.IDENTITY, new StringSerializable(tag));
        }
        return builder.build();
    }

    @Override
    public CSerializable bucketForRead(Object coordinate, BucketType bucketType) {
        return new StringSerializable((String) coordinate);
    }

    @Override
    public List<BucketType> getBucketTypes() {
        return ImmutableList.of(BucketType.IDENTITY);
    }
} public class LongOp implements Op {
    private final long val;
    public static final LongOpDeserializer DESERIALIZER = new LongOpDeserializer();

    public LongOp(long val) {
        this.val = val;
    }

    @Override
    public Op add(Op otherOp) {
        if (!(otherOp instanceof LongOp)) {
            throw new RuntimeException();
        }
        return new LongOp(val + ((LongOp) otherOp).val);
    }

    @Override
    public Op subtract(Op otherOp) {
        return new LongOp(this.val - ((LongOp) otherOp).val);
    }

    /**
     * Eclipse auto-generated
     */
    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + (int) (val ^ (val >>> 32));
        return result;
    }

    /**
     * Eclipse auto-generated
     */
    @Override
    public boolean equals(Object obj) {
        if (this == obj)
            return true;
        if (obj == null)
            return false;
        if (getClass() != obj.getClass())
            return false;
        LongOp other = (LongOp) obj;
        if (val != other.val)
            return false;
        return true;
    }

    @Override
    public byte[] serialize() {
        return Util.longToBytes(val);
    }

    public static class LongOpDeserializer implements Deserializer<LongOp> {
        @Override
        public LongOp fromBytes(byte[] bytes) {
            return new LongOp(Util.bytesToLong(bytes));
        }
    }

    public String toString() {
        return Long.toString(val);
    }

    public long getLong() {
        return val;
    }
}
import com.google.common.collect.ImmutableList;
import com.urbanairship.datacube.AsyncException;
import com.urbanairship.datacube.Batch;
import com.urbanairship.datacube.BucketType;
import com.urbanairship.datacube.DataCube;
import com.urbanairship.datacube.DataCubeIo;
import com.urbanairship.datacube.DbHarness;
import com.urbanairship.datacube.Dimension;
import com.urbanairship.datacube.ReadBuilder;
import com.urbanairship.datacube.Rollup;
import com.urbanairship.datacube.SyncLevel;
import com.urbanairship.datacube.WriteBuilder;
import com.urbanairship.datacube.bucketers.HourDayMonthBucketer;
import com.urbanairship.datacube.bucketers.StringToBytesBucketer;
import com.urbanairship.datacube.bucketers.TagsBucketer;
import com.urbanairship.datacube.ops.LongOp;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;
package com.urbanairship.datacube.tweetcountexample;



/**
 * A class that wraps the datacube operations in an intuitive interface. It offers methods for counting tweets and
 * getting various counts of interest.
 */
public class TweetCube {
    private static final Logger log = LoggerFactory.getLogger(TweetCube.class);

    private final DataCubeIo<LongOp> dataCubeIo;
    private final DataCube<LongOp> dataCube;

    /*
     *  Each dimension is some feature of a tweet.
     */
    Dimension<DateTime> timeDimension = new Dimension<DateTime>(