// Path: src/main/java/com/urbanairship/datacube/Deserializer.java
// public interface Deserializer<O extends Op> {
//     O fromBytes(byte[] bytes);
// }
// 
// Path: src/main/java/com/urbanairship/datacube/collectioninputformat/CollectionInputFormat.java
// public class CollectionInputFormat extends InputFormat<Writable, NullWritable> {
//     public static final String CONFKEY_COLLECTION = "collectioninputformat.collectionitems";
//     public static final String CONFKEY_VALCLASS = "collectioninputformat.valueclass";
// 
//     /**
//      * Stores the given collection as a configuration value. When getSplits() runs later,
//      * it will be able to deserialize the collection and use it.
//      */
//     public static <T extends Writable> void setCollection(Job job, Class<T> valueClass,
//                                                           Collection<T> collection) throws IOException {
//         setCollection(job.getConfiguration(), valueClass, collection);
//     }
// 
//     public static <T extends Writable> void setCollection(Configuration conf, Class<T> valueClass,
//                                                           Collection<T> collection) throws IOException {
//         conf.set(CONFKEY_COLLECTION, toBase64(valueClass, collection));
//         conf.set(CONFKEY_VALCLASS, valueClass.getName());
//     }
// 
//     public static <T extends Writable> String toBase64(Class<T> valueClass,
//                                                        Collection<? extends Writable> collection) throws IOException {
//         DataOutputBuffer out = new DataOutputBuffer();
//         new CollectionWritable(valueClass, collection).write(out);
//         byte[] rawBytes = Arrays.copyOf(out.getData(), out.getLength());
//         return new String(Base64.encodeBase64(rawBytes));
//     }
// 
//     @SuppressWarnings("unchecked")
//     public static <T extends Writable> Collection<T> fromBase64(Class<T> valueClass,
//                                                                 String base64) throws IOException {
//         byte[] rawBytes = Base64.decodeBase64(base64.getBytes());
//         DataInputBuffer in = new DataInputBuffer();
//         in.reset(rawBytes, rawBytes.length);
// 
//         CollectionWritable cw = new CollectionWritable();
//         cw.readFields(in);
//         return (Collection<T>) cw.getCollection();
//     }
// 
//     @Override
//     public RecordReader<Writable, NullWritable> createRecordReader(final InputSplit split, TaskAttemptContext arg1)
//             throws IOException, InterruptedException {
//         return new SingleItemRecordReader((SingleValueSplit) split);
// 
//     }
// 
//     @SuppressWarnings("unchecked")
//     @Override
//     public List<InputSplit> getSplits(JobContext ctx) throws IOException, InterruptedException {
//         String base64 = ctx.getConfiguration().get(CONFKEY_COLLECTION);
//         String valueClassName = ctx.getConfiguration().get(CONFKEY_VALCLASS);
//         Class<? extends Writable> valueClass;
//         try {
//             valueClass = (Class<? extends Writable>) Class.forName(valueClassName);
//         } catch (ClassNotFoundException e) {
//             throw new RuntimeException(e);
//         }
//         Collection<? extends Writable> collection = fromBase64(valueClass, base64);
//         List<InputSplit> splits = new ArrayList<InputSplit>(collection.size());
//         for (Writable val : collection) {
//             splits.add(new SingleValueSplit(valueClass, val));
//         }
//         return splits;
//     }
// }


import com.urbanairship.datacube.Deserializer;
import com.urbanairship.datacube.collectioninputformat.CollectionInputFormat;
import org.apache.commons.lang.ArrayUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
                log.info("Live cube is empty, running a straight copy from the backfill table");

                HBaseSnapshotter hbaseSnapshotter = new HBaseSnapshotter(conf, backfilledTableName,
                        cf, liveCubeTableName, new Path("/tmp/backfill_snapshot_hfiles"), true,
                        cubeNameKeyPrefix, Bytes.add(cubeNameKeyPrefix, fiftyBytesFF));
                return hbaseSnapshotter.runWithCheckedExceptions();
            } else {

                Job job = new Job(conf);
                backfilledHTable = new HTable(conf, backfilledTableName);

                Pair<byte[][], byte[][]> allRegionsStartAndEndKeys = backfilledHTable.getStartEndKeys();
                byte[][] internalSplitKeys = BackfillUtil.getSplitKeys(allRegionsStartAndEndKeys);
                Collection<Scan> scans = scansThisCubeOnly(cubeNameKeyPrefix, internalSplitKeys);

                // Get the scans that will cover this table, and store them in the job configuration to be used
                // as input splits.

                if (log.isDebugEnabled()) {
                    log.debug("Scans: " + scans);
                }
                CollectionInputFormat.setCollection(job, Scan.class, scans);

                // We cannot allow map tasks to retry, or we could increment the same key multiple times.
                job.getConfiguration().set("mapred.map.max.attempts", "1");

                job.setJobName("DataCube HBase backfiller");
                job.setJarByClass(HBaseBackfillMerger.class);
                job.getConfiguration().set(CONFKEY_DESERIALIZER, opDeserializer.getName());
                job.setMapperClass(HBaseBackfillMergeMapper.class);
// Path: src/main/java/com/urbanairship/datacube/Deserializer.java
// public interface Deserializer<O extends Op> {
//     O fromBytes(byte[] bytes);
// }
// 
// Path: src/main/java/com/urbanairship/datacube/collectioninputformat/CollectionInputFormat.java
// public class CollectionInputFormat extends InputFormat<Writable, NullWritable> {
//     public static final String CONFKEY_COLLECTION = "collectioninputformat.collectionitems";
//     public static final String CONFKEY_VALCLASS = "collectioninputformat.valueclass";
// 
//     /**
//      * Stores the given collection as a configuration value. When getSplits() runs later,
//      * it will be able to deserialize the collection and use it.
//      */
//     public static <T extends Writable> void setCollection(Job job, Class<T> valueClass,
//                                                           Collection<T> collection) throws IOException {
//         setCollection(job.getConfiguration(), valueClass, collection);
//     }
// 
//     public static <T extends Writable> void setCollection(Configuration conf, Class<T> valueClass,
//                                                           Collection<T> collection) throws IOException {
//         conf.set(CONFKEY_COLLECTION, toBase64(valueClass, collection));
//         conf.set(CONFKEY_VALCLASS, valueClass.getName());
//     }
// 
//     public static <T extends Writable> String toBase64(Class<T> valueClass,
//                                                        Collection<? extends Writable> collection) throws IOException {
//         DataOutputBuffer out = new DataOutputBuffer();
//         new CollectionWritable(valueClass, collection).write(out);
//         byte[] rawBytes = Arrays.copyOf(out.getData(), out.getLength());
//         return new String(Base64.encodeBase64(rawBytes));
//     }
// 
//     @SuppressWarnings("unchecked")
//     public static <T extends Writable> Collection<T> fromBase64(Class<T> valueClass,
//                                                                 String base64) throws IOException {
//         byte[] rawBytes = Base64.decodeBase64(base64.getBytes());
//         DataInputBuffer in = new DataInputBuffer();
//         in.reset(rawBytes, rawBytes.length);
// 
//         CollectionWritable cw = new CollectionWritable();
//         cw.readFields(in);
//         return (Collection<T>) cw.getCollection();
//     }
// 
//     @Override
//     public RecordReader<Writable, NullWritable> createRecordReader(final InputSplit split, TaskAttemptContext arg1)
//             throws IOException, InterruptedException {
//         return new SingleItemRecordReader((SingleValueSplit) split);
// 
//     }
// 
//     @SuppressWarnings("unchecked")
//     @Override
//     public List<InputSplit> getSplits(JobContext ctx) throws IOException, InterruptedException {
//         String base64 = ctx.getConfiguration().get(CONFKEY_COLLECTION);
//         String valueClassName = ctx.getConfiguration().get(CONFKEY_VALCLASS);
//         Class<? extends Writable> valueClass;
//         try {
//             valueClass = (Class<? extends Writable>) Class.forName(valueClassName);
//         } catch (ClassNotFoundException e) {
//             throw new RuntimeException(e);
//         }
//         Collection<? extends Writable> collection = fromBase64(valueClass, base64);
//         List<InputSplit> splits = new ArrayList<InputSplit>(collection.size());
//         for (Writable val : collection) {
//             splits.add(new SingleValueSplit(valueClass, val));
//         }
//         return splits;
//     }
// }


// Path: src/main/java/com/urbanairship/datacube/backfill/HBaseBackfillMerger.java
import com.urbanairship.datacube.Deserializer;
import com.urbanairship.datacube.collectioninputformat.CollectionInputFormat;
import org.apache.commons.lang.ArrayUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

                log.info("Live cube is empty, running a straight copy from the backfill table");

                HBaseSnapshotter hbaseSnapshotter = new HBaseSnapshotter(conf, backfilledTableName,
                        cf, liveCubeTableName, new Path("/tmp/backfill_snapshot_hfiles"), true,
                        cubeNameKeyPrefix, Bytes.add(cubeNameKeyPrefix, fiftyBytesFF));
                return hbaseSnapshotter.runWithCheckedExceptions();
            } else {

                Job job = new Job(conf);
                backfilledHTable = new HTable(conf, backfilledTableName);

                Pair<byte[][], byte[][]> allRegionsStartAndEndKeys = backfilledHTable.getStartEndKeys();
                byte[][] internalSplitKeys = BackfillUtil.getSplitKeys(allRegionsStartAndEndKeys);
                Collection<Scan> scans = scansThisCubeOnly(cubeNameKeyPrefix, internalSplitKeys);

                // Get the scans that will cover this table, and store them in the job configuration to be used
                // as input splits.

                if (log.isDebugEnabled()) {
                    log.debug("Scans: " + scans);
                }
                CollectionInputFormat.setCollection(job, Scan.class, scans);

                // We cannot allow map tasks to retry, or we could increment the same key multiple times.
                job.getConfiguration().set("mapred.map.max.attempts", "1");

                job.setJobName("DataCube HBase backfiller");
                job.setJarByClass(HBaseBackfillMerger.class);
                job.getConfiguration().set(CONFKEY_DESERIALIZER, opDeserializer.getName());
                job.setMapperClass(HBaseBackfillMergeMapper.class);
                job.setInputFormatClass(CollectionInputFormat.class);