// Path: src/main/java/com/knewton/mapreduce/constant/PropertyConstants.java
// public enum PropertyConstants {
// 
//     /**
//      * The Keyspace Name for a data run
//      */
//     KEYSPACE_NAME("com.knewton.inputformat.cassandra.keyspace"),
//     /**
//      * Column family type needs to be set if the column family type is Super.
//      */
//     COLUMN_FAMILY_TYPE("com.knewton.cassandra.cftype"),
//     /**
//      * Comparator class name for columns.
//      */
//     COLUMN_COMPARATOR("com.knewton.cassandra.column.comparator"),
//     /**
//      * A nonstandard comparator requires a subcomparator
//      */
//     COLUMN_SUBCOMPARATOR("com.knewton.cassandra.column.subcomparator"),
//     /**
//      * Partitioner for decorating keys.
//      */
//     PARTITIONER("com.knewton.partitioner"),
//     /**
//      * True if the columns are sparse, false if they're dense.
//      */
//     SPARSE_COLUMN("com.knewton.column.sparse"),
//     /**
//      * Boolean variable. Set true if compression enabled.
//      */
//     COMPRESSION_ENABLED("com.knewton.cassandra.backup.compression"),
//     /**
//      * Buffer size for decompression operation.
//      */
//     DECOMPRESS_BUFFER("com.knewton.cassandra.backup.compress.buffersize"),
//     /**
//      * Start and end dates for student events processing
//      */
//     START_DATE("com.knewton.studentevents.date.start"),
//     END_DATE("com.knewton.studentevents.date.end"),
//     /**
//      * The Column Family Name for a data run
//      */
//     COLUMN_FAMILY_NAME("com.knewton.inputformat.cassandra.columnfamily"),
//     /**
//      * MapReduce run environment. Should be one of the enum values in MREnvironment.java
//      */
//     MAPREDUCE_ENVIRONMENT("com.knewton.mapreduce.environment"),
//     /**
//      * Name of serialization factory class
//      */
//     SERIALIZATION_FACTORY_PARAMETER("com.knewton.thrift.serialization.protocol");
// 
//     public final String txt;
// 
//     private PropertyConstants(String propertyName) {
//         this.txt = propertyName;
//     }
// }


import com.knewton.mapreduce.constant.PropertyConstants;
import com.google.common.collect.Lists;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.utils.Pair;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.util.List;
import javax.annotation.Nullable;
     *         found
     */
    private List<FileStatus> cleanUpBackupDir(List<FileStatus> files) {
        boolean foundFullBackupDir = false;
        for (FileStatus nestedDirStatus : files) {
            String nestedDirName = nestedDirStatus.getPath().getName();
            if (nestedDirName.equals(FULL_PRIAM_BACKUP_DIR_NAME)) {
                foundFullBackupDir = true;
            }
        }
        if (foundFullBackupDir) {
            for (int i = 0; i < files.size(); i++) {
                FileStatus nestedDirStatus = files.get(i);
                String nestedDirName = nestedDirStatus.getPath().getName();
                if (nestedDirName.equals(INCREMENTAL_PRIAM_BACKUP_DIR_NAME)) {
                    LOG.info("Removing {}", nestedDirStatus.getPath());
                    files.remove(i);
                    i--;
                }
            }
        }
        return files;
    }

    /**
     * Configure the DataTableFilter and return it.
     *
     * @return The DataTableFilter.
     */
    private DataTablePathFilter getDataTableFilter(Configuration conf) {
// Path: src/main/java/com/knewton/mapreduce/constant/PropertyConstants.java
// public enum PropertyConstants {
// 
//     /**
//      * The Keyspace Name for a data run
//      */
//     KEYSPACE_NAME("com.knewton.inputformat.cassandra.keyspace"),
//     /**
//      * Column family type needs to be set if the column family type is Super.
//      */
//     COLUMN_FAMILY_TYPE("com.knewton.cassandra.cftype"),
//     /**
//      * Comparator class name for columns.
//      */
//     COLUMN_COMPARATOR("com.knewton.cassandra.column.comparator"),
//     /**
//      * A nonstandard comparator requires a subcomparator
//      */
//     COLUMN_SUBCOMPARATOR("com.knewton.cassandra.column.subcomparator"),
//     /**
//      * Partitioner for decorating keys.
//      */
//     PARTITIONER("com.knewton.partitioner"),
//     /**
//      * True if the columns are sparse, false if they're dense.
//      */
//     SPARSE_COLUMN("com.knewton.column.sparse"),
//     /**
//      * Boolean variable. Set true if compression enabled.
//      */
//     COMPRESSION_ENABLED("com.knewton.cassandra.backup.compression"),
//     /**
//      * Buffer size for decompression operation.
//      */
//     DECOMPRESS_BUFFER("com.knewton.cassandra.backup.compress.buffersize"),
//     /**
//      * Start and end dates for student events processing
//      */
//     START_DATE("com.knewton.studentevents.date.start"),
//     END_DATE("com.knewton.studentevents.date.end"),
//     /**
//      * The Column Family Name for a data run
//      */
//     COLUMN_FAMILY_NAME("com.knewton.inputformat.cassandra.columnfamily"),
//     /**
//      * MapReduce run environment. Should be one of the enum values in MREnvironment.java
//      */
//     MAPREDUCE_ENVIRONMENT("com.knewton.mapreduce.environment"),
//     /**
//      * Name of serialization factory class
//      */
//     SERIALIZATION_FACTORY_PARAMETER("com.knewton.thrift.serialization.protocol");
// 
//     public final String txt;
// 
//     private PropertyConstants(String propertyName) {
//         this.txt = propertyName;
//     }
// }


// Path: src/main/java/com/knewton/mapreduce/io/SSTableInputFormat.java
import com.knewton.mapreduce.constant.PropertyConstants;
import com.google.common.collect.Lists;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.utils.Pair;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.util.List;
import javax.annotation.Nullable;

     *         found
     */
    private List<FileStatus> cleanUpBackupDir(List<FileStatus> files) {
        boolean foundFullBackupDir = false;
        for (FileStatus nestedDirStatus : files) {
            String nestedDirName = nestedDirStatus.getPath().getName();
            if (nestedDirName.equals(FULL_PRIAM_BACKUP_DIR_NAME)) {
                foundFullBackupDir = true;
            }
        }
        if (foundFullBackupDir) {
            for (int i = 0; i < files.size(); i++) {
                FileStatus nestedDirStatus = files.get(i);
                String nestedDirName = nestedDirStatus.getPath().getName();
                if (nestedDirName.equals(INCREMENTAL_PRIAM_BACKUP_DIR_NAME)) {
                    LOG.info("Removing {}", nestedDirStatus.getPath());
                    files.remove(i);
                    i--;
                }
            }
        }
        return files;
    }

    /**
     * Configure the DataTableFilter and return it.
     *
     * @return The DataTableFilter.
     */
    private DataTablePathFilter getDataTableFilter(Configuration conf) {
        String operatingCF = conf.get(PropertyConstants.COLUMN_FAMILY_NAME.txt);