// 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;
/*
Copyright 2012 Urban Airship and Contributors
*/

package com.urbanairship.datacube.backfill;



/**
 * Backfilling is a 3 step process:
 * (1) snapshot the live cube data into a so-called snapshot table
 * (2) do the application-level event processing into a cube that is stored in some non-visible table
 * (3) Merge the snapshot, the backfill table, and the live cube, making changes to the live cube.
 *
 * This MR job does step 3.
 *
 * TODO InputSplit locality for mapreduce
 */
public class HBaseBackfillMerger implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(HBaseBackfillMerger.class);

    static final String CONFKEY_COLUMN_FAMILY = "hbasebackfiller.cf";
    static final String CONFKEY_LIVECUBE_TABLE_NAME = "hbasebackfiller.liveCubeTableName";
    static final String CONFKEY_SNAPSHOT_TABLE_NAME = "hbasebackfiller.snapshotTableName";
    static final String CONFKEY_BACKFILLED_TABLE_NAME = "hbasebackfiller.backfilledTableName";
    static final String CONFKEY_DESERIALIZER = "hbasebackfiller.deserializerClassName";

    private final Configuration conf;
    private final byte[] cubeNameKeyPrefix;
    private final byte[] liveCubeTableName;
    private final byte[] snapshotTableName;
    private final byte[] backfilledTableName;
    private final byte[] cf;
// 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;

/*
Copyright 2012 Urban Airship and Contributors
*/

package com.urbanairship.datacube.backfill;



/**
 * Backfilling is a 3 step process:
 * (1) snapshot the live cube data into a so-called snapshot table
 * (2) do the application-level event processing into a cube that is stored in some non-visible table
 * (3) Merge the snapshot, the backfill table, and the live cube, making changes to the live cube.
 *
 * This MR job does step 3.
 *
 * TODO InputSplit locality for mapreduce
 */
public class HBaseBackfillMerger implements Runnable {
    private static final Logger log = LoggerFactory.getLogger(HBaseBackfillMerger.class);

    static final String CONFKEY_COLUMN_FAMILY = "hbasebackfiller.cf";
    static final String CONFKEY_LIVECUBE_TABLE_NAME = "hbasebackfiller.liveCubeTableName";
    static final String CONFKEY_SNAPSHOT_TABLE_NAME = "hbasebackfiller.snapshotTableName";
    static final String CONFKEY_BACKFILLED_TABLE_NAME = "hbasebackfiller.backfilledTableName";
    static final String CONFKEY_DESERIALIZER = "hbasebackfiller.deserializerClassName";

    private final Configuration conf;
    private final byte[] cubeNameKeyPrefix;
    private final byte[] liveCubeTableName;
    private final byte[] snapshotTableName;
    private final byte[] backfilledTableName;
    private final byte[] cf;
    private final Class<? extends Deserializer<?>> opDeserializer;